Add background-thread state and toggle unit tests

New tests cover:
  - background thread states after arena_reset
  - background thread stats
  - toggle background thread on and off during parallel stress
    allocs/dallocs
  - background thread fork behavior

Add background-thread fork unit test

test_fork_background_thread (in fork.c): with the background thread enabled,
fork and assert the child comes up with it disabled yet usable (re-enable +
alloc round-trips), while the parent keeps its threads.  Placed in fork.c to
reuse its wait_for_child_exit helper and fork/WIN32 handling instead of
duplicating them in a separate file.

Makefile.in

@@ -204,7 +204,6 @@ C_UTIL_CPP_SRCS := $(srcroot)src/nstime.c $(srcroot)src/malloc_io.c
 endif
 TESTS_UNIT := \
 	$(srcroot)test/unit/a0.c \
-	$(srcroot)test/unit/arena_decay.c \
 	$(srcroot)test/unit/arena_reset.c \
 	$(srcroot)test/unit/arenas_management.c \
 	$(srcroot)test/unit/atomic.c \
@@ -272,6 +271,7 @@ TESTS_UNIT := \
 	$(srcroot)test/unit/oversize_threshold.c \
 	$(srcroot)test/unit/pa.c \
 	$(srcroot)test/unit/pac.c \
+	$(srcroot)test/unit/pac_decay.c \
 	$(srcroot)test/unit/pac_sec_integration.c \
 	$(srcroot)test/unit/pack.c \
 	$(srcroot)test/unit/pages.c \

include/jemalloc/internal/arena.h

@@ -177,12 +177,6 @@ struct arena_s {
 /* EXTERNS */
 /******************************************************************************/
 
-/*
- * When the amount of pages to be purged exceeds this amount, deferred purge
- * should happen.
- */
-#define ARENA_DEFERRED_PURGE_NPAGES_THRESHOLD UINT64_C(1024)
-
 extern ssize_t opt_dirty_decay_ms;
 extern ssize_t opt_muzzy_decay_ms;
 
@@ -213,7 +207,6 @@ void arena_stats_merge(tsdn_t *tsdn, arena_t *arena, unsigned *nthreads,
     size_t *nactive, size_t *ndirty, size_t *nmuzzy, arena_stats_t *astats,
     bin_stats_data_t *bstats, arena_stats_large_t *lstats, pac_estats_t *estats,
     hpa_shard_stats_t *hpastats);
-void arena_handle_deferred_work(tsdn_t *tsdn, arena_t *arena);
 edata_t *arena_extent_alloc_large(
     tsdn_t *tsdn, arena_t *arena, size_t usize, size_t alignment, bool zero);
 void arena_extent_dalloc_large_prep(
@@ -222,13 +215,6 @@ void arena_extent_ralloc_large_shrink(
     tsdn_t *tsdn, arena_t *arena, const edata_t *edata, size_t oldusize);
 void arena_extent_ralloc_large_expand(
     tsdn_t *tsdn, arena_t *arena, const edata_t *edata, size_t oldusize);
-bool arena_decay_ms_set(
-    tsdn_t *tsdn, arena_t *arena, extent_state_t state, ssize_t decay_ms);
-ssize_t arena_decay_ms_get(arena_t *arena, extent_state_t state);
-void    arena_decay(
-       tsdn_t *tsdn, arena_t *arena, bool is_background_thread, bool all);
-uint64_t       arena_time_until_deferred(tsdn_t *tsdn, arena_t *arena);
-void           arena_do_deferred_work(tsdn_t *tsdn, arena_t *arena);
 void           arena_reset(tsd_t *tsd, arena_t *arena);
 void           arena_destroy(tsd_t *tsd, arena_t *arena);
 cache_bin_sz_t arena_ptr_array_fill_small(tsdn_t *tsdn, arena_t *arena,

include/jemalloc/internal/arena_inlines.h

@@ -308,7 +308,7 @@ arena_decay_ticks(tsdn_t *tsdn, arena_t *arena, unsigned nticks) {
 	uint64_t      *prng_state = tsd_prng_statep_get(tsd);
 	if (unlikely(ticker_geom_ticks(decay_ticker, prng_state, nticks,
 	        tsd_reentrancy_level_get(tsd) > 0))) {
-		arena_decay(tsdn, arena, false, false);
+		pa_shard_do_deferred_work(tsdn, &arena->pa_shard, false);
 	}
 }
 

include/jemalloc/internal/background_thread.h

@@ -13,17 +13,6 @@
 #define MAX_BACKGROUND_THREAD_LIMIT MALLOCX_ARENA_LIMIT
 #define DEFAULT_NUM_BACKGROUND_THREAD 4
 
-/*
- * These exist only as a transitional state.  Eventually, deferral should be
- * part of the PAI, and each implementation can indicate wait times with more
- * specificity.
- */
-#define BACKGROUND_THREAD_HPA_INTERVAL_MAX_UNINITIALIZED (-2)
-#define BACKGROUND_THREAD_HPA_INTERVAL_MAX_DEFAULT_WHEN_ENABLED 5000
-
-#define BACKGROUND_THREAD_DEFERRED_MIN UINT64_C(0)
-#define BACKGROUND_THREAD_DEFERRED_MAX UINT64_MAX
-
 typedef enum {
 	background_thread_stopped,
 	background_thread_started,
@@ -87,6 +76,25 @@ bool background_thread_stats_read(
     tsdn_t *tsdn, background_thread_stats_t *stats);
 void background_thread_ctl_init(tsdn_t *tsdn);
 
+/*
+ * Arena-reset lifecycle bracket (owned by the bg-thread module so the
+ * started<->paused state machine is mutated only inside the module).  begin()
+ * acquires the global background_thread_lock and RETURNS WITH IT HELD;
+ * finish() releases it.  The lock intentionally spans the whole arena-reset
+ * body across the two calls.  Pass arena_ind (not an arena_t *): the destroy
+ * path frees the arena before calling finish().
+ */
+void background_thread_arena_reset_begin(tsd_t *tsd, unsigned arena_ind);
+void background_thread_arena_reset_finish(tsd_t *tsd, unsigned arena_ind);
+
+/*
+ * Serialize a rare admin operation against the background thread by holding the
+ * per-arena info mutex (have_background_thread-gated).  Used by
+ * arena_set_extent_hooks to fence pa_shard_disable_hpa.
+ */
+void background_thread_serialize_lock(tsd_t *tsd, unsigned arena_ind);
+void background_thread_serialize_unlock(tsd_t *tsd, unsigned arena_ind);
+
 #ifdef JEMALLOC_PTHREAD_CREATE_WRAPPER
 extern int pthread_create_wrapper(pthread_t *__restrict, const pthread_attr_t *,
     void *(*)(void *), void *__restrict);

include/jemalloc/internal/background_thread_inlines.h

@@ -22,12 +22,6 @@ background_thread_enabled_set(tsdn_t *tsdn, bool state) {
 	background_thread_enabled_set_impl(state);
 }
 
-JEMALLOC_ALWAYS_INLINE background_thread_info_t *
-arena_background_thread_info_get(arena_t *arena) {
-	unsigned arena_ind = arena_ind_get(arena);
-	return &background_thread_info[arena_ind % max_background_threads];
-}
-
 JEMALLOC_ALWAYS_INLINE background_thread_info_t *
 background_thread_info_get(size_t ind) {
 	return &background_thread_info[ind % max_background_threads];

include/jemalloc/internal/deferral.h

@@ -0,0 +1,30 @@
+#ifndef JEMALLOC_INTERNAL_DEFERRAL_H
+#define JEMALLOC_INTERNAL_DEFERRAL_H
+
+#include "jemalloc/internal/jemalloc_preamble.h"
+
+/*
+ * The deferred-work contract shared by the page allocators that PRODUCE
+ * deferred work (PAC decay-purge, HPA purge/hugify) and the drivers that
+ * CONSUME it (the background thread, or the application-inline fallback).
+ *
+ * There is deliberately no polymorphic interface here: with only two
+ * page allocators, a vtable's indirect calls are needless and inefficient.
+ * Dispatch is direct pac_*()/hpa_*(), so the contract is a convention
+ * documented here, not an enforced type.  The only shared code it needs is the
+ * constants below; each allocator keeps its own policy definition (functions
+ * and constants) in its own header.
+ *
+ * The convention every page allocator follows:
+ *   - <alloc>_time_until_deferred_work(): nanoseconds until its next
+ *     deferred work is due.  DEFERRED_WORK_MIN = "due now",
+ *     DEFERRED_WORK_MAX = "nothing pending"; anything between is a real
+ *     ns deadline.
+ *   - <alloc>_do_deferred_work(): perform whatever work is due now.
+ * The pa layer reports the soonest deadline across its allocators; the
+ * background thread sleeps until then, or indefinitely on DEFERRED_WORK_MAX.
+ */
+#define DEFERRED_WORK_MIN UINT64_C(0)
+#define DEFERRED_WORK_MAX UINT64_MAX
+
+#endif /* JEMALLOC_INTERNAL_DEFERRAL_H */

include/jemalloc/internal/hpa.h

@@ -13,6 +13,14 @@
 #include "jemalloc/internal/psset.h"
 #include "jemalloc/internal/sec.h"
 
+/*
+ * HPA-specific deferral interval bounds (currently unused).  Deferral tuning
+ * is per-allocator policy, so it lives in the HPA header; see deferral.h for
+ * the shared deferred-work contract.
+ */
+#define HPA_INTERVAL_MAX_UNINITIALIZED (-2)
+#define HPA_INTERVAL_MAX_DEFAULT_WHEN_ENABLED 5000
+
 typedef struct hpa_shard_nonderived_stats_s hpa_shard_nonderived_stats_t;
 struct hpa_shard_nonderived_stats_s {
 	/*

include/jemalloc/internal/pa.h

@@ -111,12 +111,6 @@ struct pa_shard_s {
 	pac_t pac;
 };
 
-static inline bool
-pa_shard_dont_decay_muzzy(pa_shard_t *shard) {
-	return ecache_npages_get(&shard->pac.ecache_muzzy) == 0
-	    && pac_decay_ms_get(&shard->pac, extent_state_muzzy) <= 0;
-}
-
 static inline ehooks_t *
 pa_shard_ehooks_get(const pa_shard_t *shard) {
 	return base_ehooks_get(shard->base);
@@ -158,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,
@@ -184,22 +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.
- *
- * Morally, this should do both PAC decay and the HPA deferred work.  For now,
- * though, the arena, background thread, and PAC modules are tightly interwoven
- * in a way that's tricky to extricate, so we only do the HPA-specific parts.
- */
 void pa_shard_set_deferral_allowed(
     tsdn_t *tsdn, pa_shard_t *shard, bool deferral_allowed);
-void     pa_shard_do_deferred_work(tsdn_t *tsdn, pa_shard_t *shard);
+/*
+ * 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, 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

@@ -13,7 +13,7 @@
 #include "san_bump.h"
 
 /*
- * Page allocator classic; an implementation of the PAI interface that:
+ * Page allocator classic (PAC), a page-level allocator that:
  * - Can be used for arenas with custom extent hooks.
  * - Can always satisfy any allocation request (including highly-fragmentary
  *   ones).
@@ -28,6 +28,14 @@ enum pac_purge_eagerness_e {
 };
 typedef enum pac_purge_eagerness_e pac_purge_eagerness_t;
 
+/*
+ * When a decay sweep would purge more than this many pages, its purge is
+ * treated as due (worth waking the background thread for) instead of left
+ * to accumulate.  PAC decay policy; the HPA defers on time intervals, not
+ * on a page count, so this constant is PAC-only.
+ */
+#define PAC_DECAY_PURGE_NPAGES_THRESHOLD UINT64_C(1024)
+
 typedef struct pac_decay_stats_s pac_decay_stats_t;
 struct pac_decay_stats_s {
 	/* Total number of purge sweeps. */
@@ -231,6 +239,41 @@ bool pac_maybe_decay_purge(tsdn_t *tsdn, pac_t *pac, decay_t *decay,
     pac_decay_stats_t *decay_stats, ecache_t *ecache,
     pac_purge_eagerness_t eagerness);
 
+/*
+ * Result of a pac_decay_deferred() pass.  Reported per decay state so
+ * pac_do_deferred_work can decide, per state, whether to notify the background
+ * thread.  npages_new is the per-state epoch backlog delta, only meaningful when
+ * the corresponding *_epoch_advanced is true.
+ */
+typedef struct pac_deferred_work_result_s pac_deferred_work_result_t;
+struct pac_deferred_work_result_s {
+	size_t dirty_npages_new;
+	size_t muzzy_npages_new;
+	bool   dirty_epoch_advanced;
+	bool   muzzy_epoch_advanced;
+};
+
+/*
+ * All deferred decay-purge work for a PAC shard: decide the eagerness from the
+ * 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, 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.
+ */
+void pac_decay_all_now(tsdn_t *tsdn, pac_t *pac, extent_state_t state);
+
 /*
  * Gets / sets the maximum amount that we'll grow an arena down the
  * grow-retained pathways (unless forced to by an allocaction request).
@@ -244,9 +287,22 @@ 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). */
+static inline bool
+pac_should_decay_muzzy(pac_t *pac) {
+	return ecache_npages_get(&pac->ecache_muzzy) != 0
+	    || 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,11 +83,6 @@ const arena_config_t arena_config_default = {
  * definition.
  */
 
-static bool arena_decay_dirty(
-    tsdn_t *tsdn, arena_t *arena, bool is_background_thread, bool all);
-static void arena_maybe_do_deferred_work(
-    tsdn_t *tsdn, arena_t *arena, decay_t *decay, size_t npages_new);
-
 /******************************************************************************/
 
 void
@@ -96,8 +91,8 @@ arena_basic_stats_merge(tsdn_t *tsdn, arena_t *arena, unsigned *nthreads,
     size_t *nactive, size_t *ndirty, size_t *nmuzzy) {
 	*nthreads += arena_nthreads_get(arena, false);
 	*dss = dss_prec_names[arena_dss_prec_get(arena)];
-	*dirty_decay_ms = arena_decay_ms_get(arena, extent_state_dirty);
+	*dirty_decay_ms = pa_decay_ms_get(&arena->pa_shard, extent_state_dirty);
-	*muzzy_decay_ms = arena_decay_ms_get(arena, extent_state_muzzy);
+	*muzzy_decay_ms = pa_decay_ms_get(&arena->pa_shard, extent_state_muzzy);
 	pa_shard_basic_stats_merge(&arena->pa_shard, nactive, ndirty, nmuzzy);
 }
 
@@ -316,34 +311,6 @@ 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.
- */
-void
-arena_handle_deferred_work(tsdn_t *tsdn, arena_t *arena) {
-	witness_assert_depth_to_rank(
-	    tsdn_witness_tsdp_get(tsdn), WITNESS_RANK_CORE, 0);
-
-	if (decay_immediately(&arena->pa_shard.pac.decay_dirty)) {
-		arena_decay_dirty(tsdn, arena, false, true);
-	}
-	arena_background_thread_inactivity_check(tsdn, arena, false);
-}
-
 static void
 arena_large_malloc_stats_update(tsdn_t *tsdn, arena_t *arena, size_t usize) {
 	cassert(config_stats);
@@ -476,188 +443,12 @@ 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(
-	    /* is_background_thread */ false);
-	return pa_decay_ms_set(
-	    tsdn, &arena->pa_shard, state, decay_ms, eagerness);
-}
-
-ssize_t
-arena_decay_ms_get(arena_t *arena, extent_state_t state) {
-	return pa_decay_ms_get(&arena->pa_shard, state);
-}
-
-static bool
-arena_decay_impl(tsdn_t *tsdn, arena_t *arena, decay_t *decay,
-    pac_decay_stats_t *decay_stats, ecache_t *ecache, bool is_background_thread,
-    bool all) {
-	if (all) {
-		malloc_mutex_lock(tsdn, &decay->mtx);
-		pac_decay_all(tsdn, &arena->pa_shard.pac, decay, decay_stats,
-		    ecache, /* fully_decay */ all);
-		malloc_mutex_unlock(tsdn, &decay->mtx);
-		return false;
-	}
-
-	if (malloc_mutex_trylock(tsdn, &decay->mtx)) {
-		/* No need to wait if another thread is in progress. */
-		return true;
-	}
-	pac_purge_eagerness_t eagerness = arena_decide_unforced_purge_eagerness(
-	    is_background_thread);
-	bool epoch_advanced = pac_maybe_decay_purge(
-	    tsdn, &arena->pa_shard.pac, decay, decay_stats, ecache, eagerness);
-	size_t npages_new JEMALLOC_CLANG_ANALYZER_SILENCE_INIT(0);
-	if (epoch_advanced) {
-		/* Backlog is updated on epoch advance. */
-		npages_new = decay_epoch_npages_delta(decay);
-	}
-	malloc_mutex_unlock(tsdn, &decay->mtx);
-
-	if (have_background_thread && background_thread_enabled()
-	    && epoch_advanced && !is_background_thread) {
-		arena_maybe_do_deferred_work(tsdn, arena, decay, npages_new);
-	}
-
-	return false;
-}
-
-static bool
-arena_decay_dirty(
-    tsdn_t *tsdn, arena_t *arena, bool is_background_thread, bool all) {
-	return arena_decay_impl(tsdn, arena, &arena->pa_shard.pac.decay_dirty,
-	    &arena->pa_shard.pac.stats->decay_dirty,
-	    &arena->pa_shard.pac.ecache_dirty, is_background_thread, all);
-}
-
-static bool
-arena_decay_muzzy(
-    tsdn_t *tsdn, arena_t *arena, bool is_background_thread, bool all) {
-	if (pa_shard_dont_decay_muzzy(&arena->pa_shard)) {
-		return false;
-	}
-	return arena_decay_impl(tsdn, arena, &arena->pa_shard.pac.decay_muzzy,
-	    &arena->pa_shard.pac.stats->decay_muzzy,
-	    &arena->pa_shard.pac.ecache_muzzy, is_background_thread, all);
-}
-
-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
-		 * as possible", including flushing any caches (for situations
-		 * like thread death, or manual purge calls).
-		 */
-		pa_shard_flush(tsdn, &arena->pa_shard);
-	}
-	if (arena_decay_dirty(tsdn, arena, is_background_thread, all)) {
-		return;
-	}
-	arena_decay_muzzy(tsdn, arena, is_background_thread, all);
-}
-
-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
-	    > ARENA_DEFERRED_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) {
-	arena_decay(tsdn, arena, true, false);
-	pa_shard_do_deferred_work(tsdn, &arena->pa_shard);
-}
-
 static void
 arena_slab_dalloc(tsdn_t *tsdn, arena_t *arena, edata_t *slab) {
 	bool deferred_work_generated = false;
 	pa_dalloc(tsdn, &arena->pa_shard, slab, &deferred_work_generated);
 	if (deferred_work_generated) {
-		arena_handle_deferred_work(tsdn, arena);
+		pa_shard_handle_deferred_work(tsdn, &arena->pa_shard);
 	}
 }
 
@@ -934,7 +725,7 @@ arena_slab_alloc(tsdn_t *tsdn, arena_t *arena, szind_t binind,
 	    /* zero */ false, guarded, &deferred_work_generated);
 
 	if (deferred_work_generated) {
-		arena_handle_deferred_work(tsdn, arena);
+		pa_shard_handle_deferred_work(tsdn, &arena->pa_shard);
 	}
 
 	if (slab == NULL) {
@@ -1585,17 +1376,17 @@ arena_get_ehooks(const arena_t *arena) {
 extent_hooks_t *
 arena_set_extent_hooks(
     tsd_t *tsd, arena_t *arena, extent_hooks_t *extent_hooks) {
-	background_thread_info_t *info;
+	/*
-	if (have_background_thread) {
+	 * Serialize disabling the HPA against the background thread via the
-		info = arena_background_thread_info_get(arena);
+	 * module-owned bracket (have_background_thread-gated internally) rather
-		malloc_mutex_lock(tsd_tsdn(tsd), &info->mtx);
+	 * than reaching into info->mtx directly.
-	}
+	 */
+	unsigned arena_ind = arena_ind_get(arena);
+	background_thread_serialize_lock(tsd, arena_ind);
 	/* No using the HPA now that we have the custom hooks. */
 	pa_shard_disable_hpa(tsd_tsdn(tsd), &arena->pa_shard);
 	extent_hooks_t *ret = base_extent_hooks_set(arena->base, extent_hooks);
-	if (have_background_thread) {
+	background_thread_serialize_unlock(tsd, arena_ind);
-		malloc_mutex_unlock(tsd_tsdn(tsd), &info->mtx);
-	}
 
 	return ret;
 }
@@ -1828,13 +1619,13 @@ arena_create_huge_arena(tsd_t *tsd, unsigned ind) {
 	 */
 	if (!background_thread_enabled()
 	    && arena_dirty_decay_ms_default_get() > 0) {
-		arena_decay_ms_set(
+		pa_decay_ms_set(
-		    tsd_tsdn(tsd), huge_arena, extent_state_dirty, 0);
+		    tsd_tsdn(tsd), &huge_arena->pa_shard, extent_state_dirty, 0);
 	}
 	if (!background_thread_enabled()
 	    && arena_muzzy_decay_ms_default_get() > 0) {
-		arena_decay_ms_set(
+		pa_decay_ms_set(
-		    tsd_tsdn(tsd), huge_arena, extent_state_muzzy, 0);
+		    tsd_tsdn(tsd), &huge_arena->pa_shard, extent_state_muzzy, 0);
 	}
 
 	return huge_arena;

src/arenas_management.c

@@ -202,8 +202,8 @@ arena_migrate(tsd_t *tsd, arena_t *oldarena, arena_t *newarena) {
 		 * Purge if the old arena has no associated threads anymore and
 		 * no background threads.
 		 */
-		arena_decay(tsd_tsdn(tsd), oldarena,
+		pa_shard_flush(tsd_tsdn(tsd), &oldarena->pa_shard,
-		    /* is_background_thread */ false, /* all */ true);
+		    /* all */ true);
 	}
 }
 

src/background_thread.c

@@ -6,6 +6,7 @@
 #include "jemalloc/internal/background_thread.h"
 #include "jemalloc/internal/background_thread_inlines.h"
 #include "jemalloc/internal/ctl.h"
+#include "jemalloc/internal/deferral.h"
 #include "jemalloc/internal/jemalloc_internal_inlines_a.h"
 #include "jemalloc/internal/malloc_io.h"
 #include "jemalloc/internal/mutex.h"
@@ -33,6 +34,72 @@ size_t     max_background_threads;
 /* Thread info per-index. */
 background_thread_info_t *background_thread_info;
 
+/******************************************************************************/
+/*
+ * Config-independent lifecycle/state-ownership helpers.  Defined
+ * unconditionally and gated at runtime on have_background_thread, so callers in
+ * ctl.c / arena.c never touch background_thread_lock or info->state directly;
+ * they compile to runtime no-ops when !have_background_thread.
+ */
+
+void
+background_thread_arena_reset_begin(tsd_t *tsd, unsigned arena_ind) {
+	/* Temporarily disable the background thread during arena reset. */
+	if (have_background_thread) {
+		/*
+		 * Hold background_thread_lock across the whole arena-reset
+		 * body (acquired here, released in _finish) so a concurrent
+		 * background_threads_enable() cannot start a background
+		 * thread mid-reset.  This must happen whenever the feature
+		 * is compiled in (have_background_thread), even when the
+		 * thread is not currently enabled; the state transition
+		 * below is separately gated on background_thread_enabled().
+		 */
+		malloc_mutex_lock(tsd_tsdn(tsd), &background_thread_lock);
+		if (background_thread_enabled()) {
+			background_thread_info_t *info =
+			    background_thread_info_get(arena_ind);
+			assert(info->state == background_thread_started);
+			malloc_mutex_lock(tsd_tsdn(tsd), &info->mtx);
+			info->state = background_thread_paused;
+			malloc_mutex_unlock(tsd_tsdn(tsd), &info->mtx);
+		}
+	}
+}
+
+void
+background_thread_arena_reset_finish(tsd_t *tsd, unsigned arena_ind) {
+	if (have_background_thread) {
+		if (background_thread_enabled()) {
+			background_thread_info_t *info =
+			    background_thread_info_get(arena_ind);
+			assert(info->state == background_thread_paused);
+			malloc_mutex_lock(tsd_tsdn(tsd), &info->mtx);
+			info->state = background_thread_started;
+			malloc_mutex_unlock(tsd_tsdn(tsd), &info->mtx);
+		}
+		malloc_mutex_unlock(tsd_tsdn(tsd), &background_thread_lock);
+	}
+}
+
+void
+background_thread_serialize_lock(tsd_t *tsd, unsigned arena_ind) {
+	if (have_background_thread) {
+		background_thread_info_t *info =
+		    background_thread_info_get(arena_ind);
+		malloc_mutex_lock(tsd_tsdn(tsd), &info->mtx);
+	}
+}
+
+void
+background_thread_serialize_unlock(tsd_t *tsd, unsigned arena_ind) {
+	if (have_background_thread) {
+		background_thread_info_t *info =
+		    background_thread_info_get(arena_ind);
+		malloc_mutex_unlock(tsd_tsdn(tsd), &info->mtx);
+	}
+}
+
 /******************************************************************************/
 
 #ifdef JEMALLOC_PTHREAD_CREATE_WRAPPER
@@ -298,7 +365,7 @@ background_thread_pause_check(tsdn_t *tsdn, background_thread_info_t *info) {
 static inline void
 background_work_sleep_once(
     tsdn_t *tsdn, background_thread_info_t *info, unsigned ind) {
-	uint64_t ns_until_deferred = BACKGROUND_THREAD_DEFERRED_MAX;
+	uint64_t ns_until_deferred = DEFERRED_WORK_MAX;
 	unsigned narenas = narenas_total_get();
 	bool     slept_indefinitely = background_thread_indefinite_sleep(info);
 
@@ -313,7 +380,8 @@ background_work_sleep_once(
 		 * work that caused this thread to wake up is scheduled for.
 		 */
 		if (!slept_indefinitely) {
-			arena_do_deferred_work(tsdn, arena);
+			pa_shard_do_deferred_work(tsdn, &arena->pa_shard,
+			    /* is_background_thread */ true);
 		}
 		if (ns_until_deferred <= BACKGROUND_THREAD_MIN_INTERVAL_NS) {
 			/* Min interval will be used. */
@@ -327,7 +395,7 @@ background_work_sleep_once(
 	}
 
 	uint64_t sleep_ns;
-	if (ns_until_deferred == BACKGROUND_THREAD_DEFERRED_MAX) {
+	if (ns_until_deferred == DEFERRED_WORK_MAX) {
 		sleep_ns = BACKGROUND_THREAD_INDEFINITE_SLEEP;
 	} else {
 		sleep_ns = (ns_until_deferred

src/ctl.c

@@ -2624,7 +2624,7 @@ thread_idle_ctl(tsd_t *tsd, const size_t *mib, size_t miblen, void *oldp,
 	if (opt_narenas > ncpus * 2) {
 		arena_t *arena = arena_choose(tsd, NULL);
 		if (arena != NULL) {
-			arena_decay(tsd_tsdn(tsd), arena, false, true);
+			pa_shard_flush(tsd_tsdn(tsd), &arena->pa_shard, true);
 		}
 		/*
 		 * The missing arena case is not actually an error; a thread
@@ -2746,7 +2746,13 @@ arena_i_decay(tsdn_t *tsdn, unsigned arena_ind, bool all) {
 
 	for (unsigned i = 0; i < count; i++) {
 		if (tarenas[i] != NULL) {
-			arena_decay(tsdn, tarenas[i], false, all);
+			if (all) {
+				pa_shard_flush(tsdn, &tarenas[i]->pa_shard,
+				    true);
+			} else {
+				pa_shard_do_deferred_work(
+				    tsdn, &tarenas[i]->pa_shard, false);
+			}
 		}
 	}
 }
@@ -2798,37 +2804,6 @@ arena_i_reset_destroy_helper(tsd_t *tsd, const size_t *mib, size_t miblen,
 	return ret;
 }
 
-static void
-arena_reset_prepare_background_thread(tsd_t *tsd, unsigned arena_ind) {
-	/* Temporarily disable the background thread during arena reset. */
-	if (have_background_thread) {
-		malloc_mutex_lock(tsd_tsdn(tsd), &background_thread_lock);
-		if (background_thread_enabled()) {
-			background_thread_info_t *info =
-			    background_thread_info_get(arena_ind);
-			assert(info->state == background_thread_started);
-			malloc_mutex_lock(tsd_tsdn(tsd), &info->mtx);
-			info->state = background_thread_paused;
-			malloc_mutex_unlock(tsd_tsdn(tsd), &info->mtx);
-		}
-	}
-}
-
-static void
-arena_reset_finish_background_thread(tsd_t *tsd, unsigned arena_ind) {
-	if (have_background_thread) {
-		if (background_thread_enabled()) {
-			background_thread_info_t *info =
-			    background_thread_info_get(arena_ind);
-			assert(info->state == background_thread_paused);
-			malloc_mutex_lock(tsd_tsdn(tsd), &info->mtx);
-			info->state = background_thread_started;
-			malloc_mutex_unlock(tsd_tsdn(tsd), &info->mtx);
-		}
-		malloc_mutex_unlock(tsd_tsdn(tsd), &background_thread_lock);
-	}
-}
-
 static int
 arena_i_reset_ctl(tsd_t *tsd, const size_t *mib, size_t miblen, void *oldp,
     size_t *oldlenp, void *newp, size_t newlen) {
@@ -2842,9 +2817,9 @@ arena_i_reset_ctl(tsd_t *tsd, const size_t *mib, size_t miblen, void *oldp,
 		return ret;
 	}
 
-	arena_reset_prepare_background_thread(tsd, arena_ind);
+	background_thread_arena_reset_begin(tsd, arena_ind);
 	arena_reset(tsd, arena);
-	arena_reset_finish_background_thread(tsd, arena_ind);
+	background_thread_arena_reset_finish(tsd, arena_ind);
 
 	return ret;
 }
@@ -2871,10 +2846,10 @@ arena_i_destroy_ctl(tsd_t *tsd, const size_t *mib, size_t miblen, void *oldp,
 		goto label_return;
 	}
 
-	arena_reset_prepare_background_thread(tsd, arena_ind);
+	background_thread_arena_reset_begin(tsd, arena_ind);
 	/* Merge stats after resetting and purging arena. */
 	arena_reset(tsd, arena);
-	arena_decay(tsd_tsdn(tsd), arena, false, true);
+	pa_shard_flush(tsd_tsdn(tsd), &arena->pa_shard, true);
 	ctl_darena = arenas_i(MALLCTL_ARENAS_DESTROYED);
 	ctl_darena->initialized = true;
 	ctl_arena_refresh(tsd_tsdn(tsd), arena, ctl_darena, arena_ind, true);
@@ -2885,7 +2860,7 @@ arena_i_destroy_ctl(tsd_t *tsd, const size_t *mib, size_t miblen, void *oldp,
 	/* Record arena index for later recycling via arenas.create. */
 	ql_elm_new(ctl_arena, destroyed_link);
 	ql_tail_insert(&ctl_arenas->destroyed, ctl_arena, destroyed_link);
-	arena_reset_finish_background_thread(tsd, arena_ind);
+	background_thread_arena_reset_finish(tsd, arena_ind);
 
 	assert(ret == 0);
 label_return:
@@ -3005,7 +2980,7 @@ arena_i_decay_ms_ctl_impl(tsd_t *tsd, const size_t *mib, size_t miblen,
 	}
 
 	extent_state_t state = dirty ? extent_state_dirty : extent_state_muzzy;
-	ssize_t oldval = arena_decay_ms_get(arena, state);
+	ssize_t oldval = pa_decay_ms_get(&arena->pa_shard, state);
 	ret = ctl_read(oldp, oldlenp, &oldval, sizeof(oldval));
 	if (ret != 0 || newp == NULL) {
 		return ret;
@@ -3013,7 +2988,8 @@ arena_i_decay_ms_ctl_impl(tsd_t *tsd, const size_t *mib, size_t miblen,
 
 	ssize_t newval;
 	ret = ctl_write(&newval, sizeof(newval), newp, newlen);
-	if (ret == 0 && arena_decay_ms_set(tsd_tsdn(tsd), arena, state, newval)) {
+	if (ret == 0 && pa_decay_ms_set(tsd_tsdn(tsd), &arena->pa_shard,
+	    state, newval)) {
 		ret = EFAULT;
 	}
 	return ret;

src/hpa.c

@@ -1,6 +1,7 @@
 #include "jemalloc/internal/jemalloc_preamble.h"
 
 #include "jemalloc/internal/background_thread.h"
+#include "jemalloc/internal/deferral.h"
 #include "jemalloc/internal/hpa.h"
 #include "jemalloc/internal/hpa_utils.h"
 #include "jemalloc/internal/jemalloc_probe.h"
@@ -1081,7 +1082,7 @@ hpa_dalloc(tsdn_t *tsdn, hpa_shard_t *shard, edata_t *edata,
  */
 uint64_t
 hpa_time_until_deferred_work(tsdn_t *tsdn, hpa_shard_t *shard) {
-	uint64_t time_ns = BACKGROUND_THREAD_DEFERRED_MAX;
+	uint64_t time_ns = DEFERRED_WORK_MAX;
 
 	malloc_mutex_lock(tsdn, &shard->mtx);
 
@@ -1101,7 +1102,7 @@ hpa_time_until_deferred_work(tsdn_t *tsdn, hpa_shard_t *shard) {
 			time_ns *= 1000 * 1000;
 		} else {
 			malloc_mutex_unlock(tsdn, &shard->mtx);
-			return BACKGROUND_THREAD_DEFERRED_MIN;
+			return DEFERRED_WORK_MIN;
 		}
 	}
 
@@ -1112,7 +1113,7 @@ hpa_time_until_deferred_work(tsdn_t *tsdn, hpa_shard_t *shard) {
 		 */
 		if (shard->stats.npurge_passes == 0) {
 			malloc_mutex_unlock(tsdn, &shard->mtx);
-			return BACKGROUND_THREAD_DEFERRED_MIN;
+			return DEFERRED_WORK_MIN;
 		}
 		uint64_t since_last_purge_ms = shard->central->hooks.ms_since(
 		    &shard->last_purge);
@@ -1127,7 +1128,7 @@ hpa_time_until_deferred_work(tsdn_t *tsdn, hpa_shard_t *shard) {
 				time_ns = until_purge_ns;
 			}
 		} else {
-			time_ns = BACKGROUND_THREAD_DEFERRED_MIN;
+			time_ns = DEFERRED_WORK_MIN;
 		}
 	}
 	malloc_mutex_unlock(tsdn, &shard->mtx);

src/large.c

@@ -78,7 +78,7 @@ large_ralloc_no_move_shrink(tsdn_t *tsdn, edata_t *edata, size_t usize) {
 		return true;
 	}
 	if (deferred_work_generated) {
-		arena_handle_deferred_work(tsdn, arena);
+		pa_shard_handle_deferred_work(tsdn, &arena->pa_shard);
 	}
 	arena_extent_ralloc_large_shrink(tsdn, arena, edata, old_usize);
 
@@ -101,7 +101,7 @@ large_ralloc_no_move_expand(
 	    szind, zero, &deferred_work_generated);
 
 	if (deferred_work_generated) {
-		arena_handle_deferred_work(tsdn, arena);
+		pa_shard_handle_deferred_work(tsdn, &arena->pa_shard);
 	}
 
 	if (err) {
@@ -247,7 +247,7 @@ large_dalloc_finish_impl(tsdn_t *tsdn, arena_t *arena, edata_t *edata) {
 	bool deferred_work_generated = false;
 	pa_dalloc(tsdn, &arena->pa_shard, edata, &deferred_work_generated);
 	if (deferred_work_generated) {
-		arena_handle_deferred_work(tsdn, arena);
+		pa_shard_handle_deferred_work(tsdn, &arena->pa_shard);
 	}
 }
 

src/pa.c

@@ -1,6 +1,8 @@
 #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"
 
@@ -89,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
@@ -231,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,8 +258,28 @@ pa_shard_set_deferral_allowed(
 }
 
 void
-pa_shard_do_deferred_work(tsdn_t *tsdn, pa_shard_t *shard) {
+pa_shard_handle_deferred_work(tsdn_t *tsdn, pa_shard_t *shard) {
-	if (pa_shard_uses_hpa(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, bool is_background_thread) {
+	pac_do_deferred_work(tsdn, &shard->pac, is_background_thread);
+	/*
+	 * Application threads self-throttle HPA deferred work inline from their
+	 * own alloc/dalloc path (hpa_shard_maybe_do_deferred_work, capped), so
+	 * only drive it (forced, uncapped) from here on the background thread.
+	 */
+	if (is_background_thread && pa_shard_uses_hpa(shard)) {
 		hpa_shard_do_deferred_work(tsdn, &shard->hpa);
 	}
 }
@@ -264,7 +292,7 @@ pa_shard_do_deferred_work(tsdn_t *tsdn, pa_shard_t *shard) {
 uint64_t
 pa_shard_time_until_deferred_work(tsdn_t *tsdn, pa_shard_t *shard) {
 	uint64_t time = pac_time_until_deferred_work(tsdn, &shard->pac);
-	if (time == BACKGROUND_THREAD_DEFERRED_MIN) {
+	if (time == DEFERRED_WORK_MIN) {
 		return time;
 	}
 

src/pac.c

@@ -2,6 +2,8 @@
 
 #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"
 #include "jemalloc/internal/san.h"
@@ -482,10 +484,10 @@ static inline uint64_t
 pac_ns_until_purge(tsdn_t *tsdn, decay_t *decay, size_t npages) {
 	if (malloc_mutex_trylock(tsdn, &decay->mtx)) {
 		/* Use minimal interval if decay is contended. */
-		return BACKGROUND_THREAD_DEFERRED_MIN;
+		return DEFERRED_WORK_MIN;
 	}
 	uint64_t result = decay_ns_until_purge(
-	    decay, npages, ARENA_DEFERRED_PURGE_NPAGES_THRESHOLD);
+	    decay, npages, PAC_DECAY_PURGE_NPAGES_THRESHOLD);
 
 	malloc_mutex_unlock(tsdn, &decay->mtx);
 	return result;
@@ -497,7 +499,7 @@ pac_time_until_deferred_work(tsdn_t *tsdn, pac_t *pac) {
 
 	time = pac_ns_until_purge(
 	    tsdn, &pac->decay_dirty, ecache_npages_get(&pac->ecache_dirty));
-	if (time == BACKGROUND_THREAD_DEFERRED_MIN) {
+	if (time == DEFERRED_WORK_MIN) {
 		return time;
 	}
 
@@ -779,9 +781,238 @@ pac_maybe_decay_purge(tsdn_t *tsdn, pac_t *pac, decay_t *decay,
 	return epoch_advanced;
 }
 
+/*
+ * Run the deferred (non-forced) decay-purge for a single decay state, taking
+ * decay->mtx via trylock.  Sets *contended when the lock could not be acquired
+ * (in which case the return is false and *npages_new is untouched).  Returns
+ * whether the epoch advanced; when it did, *npages_new is set to the fresh
+ * backlog delta.
+ */
+static bool
+pac_decay_deferred_one(tsdn_t *tsdn, pac_t *pac, decay_t *decay,
+    pac_decay_stats_t *decay_stats, ecache_t *ecache,
+    pac_purge_eagerness_t eagerness, bool *contended, size_t *npages_new) {
+	if (malloc_mutex_trylock(tsdn, &decay->mtx)) {
+		/* No need to wait if another thread is in progress. */
+		*contended = true;
+		return false;
+	}
+	*contended = false;
+	bool epoch_advanced = pac_maybe_decay_purge(
+	    tsdn, pac, decay, decay_stats, ecache, eagerness);
+	if (epoch_advanced) {
+		/* Backlog is updated on epoch advance. */
+		*npages_new = decay_epoch_npages_delta(decay);
+	}
+	malloc_mutex_unlock(tsdn, &decay->mtx);
+	return epoch_advanced;
+}
+
+/*
+ * 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));
+
+	bool contended;
+	result->dirty_epoch_advanced = pac_decay_deferred_one(tsdn, pac,
+	    &pac->decay_dirty, &pac->stats->decay_dirty, &pac->ecache_dirty,
+	    eagerness, &contended, &result->dirty_npages_new);
+	if (contended) {
+		/* When dirty decay is contended, don't wait on muzzy. */
+		return;
+	}
+
+	if (!pac_should_decay_muzzy(pac)) {
+		return;
+	}
+	result->muzzy_epoch_advanced = pac_decay_deferred_one(tsdn, pac,
+	    &pac->decay_muzzy, &pac->stats->decay_muzzy, &pac->ecache_muzzy,
+	    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;
+	pac_decay_stats_t *decay_stats;
+	ecache_t          *ecache;
+	pac_decay_data_get(pac, state, &decay, &decay_stats, &ecache);
+
+	malloc_mutex_lock(tsdn, &decay->mtx);
+	pac_decay_all(
+	    tsdn, pac, decay, decay_stats, ecache, /* fully_decay */ true);
+	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;
@@ -810,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);
 

src/tcache.c

@@ -1234,17 +1234,18 @@ tcache_destroy(tsd_t *tsd, tcache_t *tcache, bool tsd_tcache) {
 	 * tsd).  Manually trigger decay to avoid pathological cases.  Also
 	 * include arena 0 because the tcache array is allocated from it.
 	 */
-	arena_decay(
+	pa_shard_do_deferred_work(tsd_tsdn(tsd),
-	    tsd_tsdn(tsd), arena_get(tsd_tsdn(tsd), 0, false), false, false);
+	    &arena_get(tsd_tsdn(tsd), 0, false)->pa_shard,
+	    /* is_background_thread */ false);
 
 	if (arena_nthreads_get(arena, false) == 0
 	    && !background_thread_enabled()) {
 		/* Force purging when no threads assigned to the arena anymore. */
-		arena_decay(tsd_tsdn(tsd), arena,
+		pa_shard_flush(tsd_tsdn(tsd), &arena->pa_shard,
-		    /* is_background_thread */ false, /* all */ true);
+		    /* all */ true);
 	} else {
-		arena_decay(tsd_tsdn(tsd), arena,
+		pa_shard_do_deferred_work(tsd_tsdn(tsd), &arena->pa_shard,
-		    /* is_background_thread */ false, /* all */ false);
+		    /* is_background_thread */ false);
 	}
 }
 

test/unit/background_thread.c

@@ -107,9 +107,88 @@ TEST_BEGIN(test_background_thread_running) {
 }
 TEST_END
 
+TEST_BEGIN(test_background_thread_arena_reset) {
+	test_skip_if(!have_background_thread);
+
+	test_switch_background_thread_ctl(true);
+
+	unsigned arena_ind;
+	size_t   sz = sizeof(arena_ind);
+	expect_d_eq(mallctl("arenas.create", (void *)&arena_ind, &sz, NULL, 0),
+	    0, "Unexpected arenas.create failure");
+	void *p = mallocx(PAGE,
+	    MALLOCX_TCACHE_NONE | MALLOCX_ARENA(arena_ind));
+	expect_ptr_not_null(p, "Unexpected mallocx failure");
+
+	/*
+	 * Resetting an arena takes its background thread through
+	 * started -> paused -> started (background_thread_arena_reset_begin/
+	 * finish).  The reset frees p, so we must not touch it afterwards.
+	 */
+	size_t mib[3];
+	size_t miblen = sizeof(mib) / sizeof(mib[0]);
+	expect_d_eq(mallctlnametomib("arena.0.reset", mib, &miblen), 0,
+	    "Unexpected mallctlnametomib failure");
+	mib[1] = arena_ind;
+	expect_d_eq(mallctlbymib(mib, miblen, NULL, NULL, NULL, 0), 0,
+	    "Unexpected arena reset failure");
+
+#if defined(JEMALLOC_BACKGROUND_THREAD)
+	background_thread_info_t *info = background_thread_info_get(arena_ind);
+	tsd_t                    *tsd = tsd_fetch();
+	malloc_mutex_lock(tsd_tsdn(tsd), &info->mtx);
+	background_thread_state_t st = info->state;
+	malloc_mutex_unlock(tsd_tsdn(tsd), &info->mtx);
+	expect_d_eq((int)st, (int)background_thread_started,
+	    "Arena reset must leave the background thread started, not paused");
+#endif
+	expect_zu_gt(n_background_threads, 0,
+	    "Arena reset must not lose background threads");
+
+	test_switch_background_thread_ctl(false);
+}
+TEST_END
+
+TEST_BEGIN(test_background_thread_stats_ctl) {
+	test_skip_if(!have_background_thread);
+	test_skip_if(!config_stats);
+
+	test_switch_background_thread_ctl(true);
+
+	uint64_t epoch = 1;
+	size_t   sz = sizeof(epoch);
+	expect_d_eq(mallctl("epoch", (void *)&epoch, &sz, (void *)&epoch,
+	                sizeof(epoch)),
+	    0, "Unexpected epoch mallctl failure");
+
+	size_t num_threads;
+	sz = sizeof(num_threads);
+	expect_d_eq(mallctl("stats.background_thread.num_threads",
+	                (void *)&num_threads, &sz, NULL, 0),
+	    0, "Unexpected stats.background_thread.num_threads failure");
+	expect_zu_eq(num_threads, n_background_threads,
+	    "Reported num_threads should match n_background_threads");
+
+	uint64_t num_runs;
+	sz = sizeof(num_runs);
+	expect_d_eq(mallctl("stats.background_thread.num_runs",
+	                (void *)&num_runs, &sz, NULL, 0),
+	    0, "Unexpected stats.background_thread.num_runs failure");
+
+	uint64_t run_interval;
+	sz = sizeof(run_interval);
+	expect_d_eq(mallctl("stats.background_thread.run_interval",
+	                (void *)&run_interval, &sz, NULL, 0),
+	    0, "Unexpected stats.background_thread.run_interval failure");
+
+	test_switch_background_thread_ctl(false);
+}
+TEST_END
+
 int
 main(void) {
 	/* Background_thread creation tests reentrancy naturally. */
-	return test_no_reentrancy(
+	return test_no_reentrancy(test_background_thread_ctl,
-	    test_background_thread_ctl, test_background_thread_running);
+	    test_background_thread_running, test_background_thread_arena_reset,
+	    test_background_thread_stats_ctl);
 }

test/unit/background_thread_enable.c

@@ -96,7 +96,52 @@ TEST_BEGIN(test_max_background_threads) {
 }
 TEST_END
 
+static atomic_b_t stress_stop;
+
+static void
+set_background_thread(bool enable) {
+	size_t sz = sizeof(bool);
+	expect_d_eq(mallctl("background_thread", NULL, NULL, &enable, sz), 0,
+	    "Failed to set background_thread");
+}
+
+static void *
+stress_worker(void *arg) {
+	(void)arg;
+	while (!atomic_load_b(&stress_stop, ATOMIC_RELAXED)) {
+		void *p = mallocx(PAGE, MALLOCX_TCACHE_NONE);
+		if (p != NULL) {
+			dallocx(p, MALLOCX_TCACHE_NONE);
+		}
+	}
+	return NULL;
+}
+
+TEST_BEGIN(test_toggle_stress_with_concurrent_alloc) {
+	test_skip_if(!have_background_thread);
+
+	atomic_store_b(&stress_stop, false, ATOMIC_RELAXED);
+	thd_t thd;
+	thd_create(&thd, stress_worker, NULL);
+
+	for (unsigned i = 0; i < 200; i++) {
+		set_background_thread(true);
+		expect_zu_gt(n_background_threads, 0,
+		    "Background threads should be non-zero after enable "
+		    "(cycle %u)", i);
+		set_background_thread(false);
+		expect_zu_eq(n_background_threads, 0,
+		    "Background threads should be zero after disable "
+		    "(cycle %u)", i);
+	}
+
+	atomic_store_b(&stress_stop, true, ATOMIC_RELAXED);
+	thd_join(thd, NULL);
+}
+TEST_END
+
 int
 main(void) {
-	return test_no_reentrancy(test_deferred, test_max_background_threads);
+	return test_no_reentrancy(test_deferred, test_max_background_threads,
+	    test_toggle_stress_with_concurrent_alloc);
 }

test/unit/decay.c

@@ -273,9 +273,67 @@ TEST_BEGIN(test_decay_ns_until_purge) {
 }
 TEST_END
 
+/*
+ * Branch coverage for the decay-level math the early-wake path
+ * (pac_decay_should_wake_early) relies on, exercised here at the public decay
+ * seam with explicit nstime rather than via real-thread timing.
+ */
+TEST_BEGIN(test_decay_early_wake_branches) {
+	decay_t decay;
+	memset(&decay, 0, sizeof(decay));
+
+	nstime_t curtime;
+	nstime_init(&curtime, 0);
+
+	uint64_t decay_ms = 1000;
+	nstime_t decay_nstime;
+	nstime_init(&decay_nstime, decay_ms * 1000 * 1000);
+	expect_false(decay_init(&decay, &curtime, (ssize_t)decay_ms),
+	    "Failed to initialize decay");
+
+	size_t new_pages = 100;
+
+	/*
+	 * Parity trap: with zero new pages the accumulation contributes
+	 * nothing, so decay_npages_purge_in must return 0 regardless of the
+	 * remaining-sleep window.  (The early-wake THRESHOLD compare on the
+	 * EXISTING backlog still runs in pac_decay_should_wake_early; that is
+	 * deliberately kept and documented in src/pac.c.)
+	 */
+	nstime_t window;
+	nstime_copy(&window, &decay_nstime);
+	expect_u64_eq(decay_npages_purge_in(&decay, &window, 0), 0,
+	    "Zero new pages must contribute zero to the purge estimate");
+
+	/*
+	 * A remaining-sleep window at/after the full decay period caps the
+	 * estimate at the new-page count (all of them will have decayed).
+	 */
+	expect_u64_eq(decay_npages_purge_in(&decay, &window, new_pages),
+	    new_pages, "All new pages should be due within the decay window");
+
+	nstime_t beyond;
+	nstime_copy(&beyond, &decay_nstime);
+	nstime_add(&beyond, &decay_nstime);
+	expect_u64_eq(decay_npages_purge_in(&decay, &beyond, new_pages),
+	    new_pages,
+	    "Estimate must cap at the new-page count beyond the decay window");
+
+	/*
+	 * No-backlog branch: with a zero threshold, decay_ns_until_purge is
+	 * unbounded (nothing schedulable).  This mirrors the indefinite-sleep
+	 * case.
+	 */
+	expect_u64_eq(decay_ns_until_purge(&decay, 0, 0),
+	    DECAY_UNBOUNDED_TIME_TO_PURGE,
+	    "Zero threshold must yield an unbounded time-to-purge");
+}
+TEST_END
+
 int
 main(void) {
 	return test(test_decay_init, test_decay_ms_valid,
 	    test_decay_npages_purge_in, test_decay_maybe_advance_epoch,
-	    test_decay_empty, test_decay, test_decay_ns_until_purge);
+	    test_decay_empty, test_decay, test_decay_ns_until_purge,
+	    test_decay_early_wake_branches);
 }

test/unit/fork.c

@@ -568,10 +568,98 @@ TEST_BEGIN(test_fork_postfork_double_fork) {
 }
 TEST_END
 
+#ifndef _WIN32
+static bool
+get_background_thread_enabled(void) {
+	bool   enabled = false;
+	size_t sz = sizeof(enabled);
+	if (mallctl("background_thread", (void *)&enabled, &sz, NULL, 0) != 0) {
+		return false;
+	}
+	return enabled;
+}
+
+static bool
+set_background_thread_enabled(bool enabled) {
+	return mallctl("background_thread", NULL, NULL, &enabled,
+	           sizeof(enabled)) == 0;
+}
+#endif
+
+TEST_BEGIN(test_fork_background_thread) {
+#ifndef _WIN32
+	test_skip_if(!have_background_thread);
+
+	/* Enable the background thread in the parent. */
+	bool enabled = true;
+	expect_d_eq(mallctl("background_thread", NULL, NULL, &enabled,
+	                sizeof(enabled)),
+	    0, "Unexpected mallctl() failure");
+	expect_zu_gt(n_background_threads, 0,
+	    "Number of background threads should be non zero after enabling.");
+
+	pid_t pid = fork();
+	if (pid == -1) {
+		test_fail("Unexpected fork() failure");
+	} else if (pid == 0) {
+		/*
+		 * Child: postfork_child disables the background thread, but the
+		 * allocator must remain usable and re-init capable.
+		 */
+		if (get_background_thread_enabled()) {
+			/* Should report disabled in the child. */
+			_exit(1);
+		}
+
+		/* Several malloc/free round-trips to confirm usability. */
+		for (unsigned i = 0; i < 16; i++) {
+			void *p = mallocx(64, MALLOCX_TCACHE_NONE);
+			if (p == NULL) {
+				_exit(2);
+			}
+			dallocx(p, MALLOCX_TCACHE_NONE);
+		}
+
+		void *q = malloc(128);
+		if (q == NULL) {
+			_exit(3);
+		}
+		free(q);
+
+		/* Re-enable the background thread in the child. */
+		if (!set_background_thread_enabled(true)) {
+			_exit(4);
+		}
+		if (!get_background_thread_enabled()) {
+			_exit(5);
+		}
+
+		/* And confirm allocation still works afterwards. */
+		void *r = mallocx(256, MALLOCX_TCACHE_NONE);
+		if (r == NULL) {
+			_exit(6);
+		}
+		dallocx(r, MALLOCX_TCACHE_NONE);
+
+		_exit(0);
+	} else {
+		/* Parent: keeps its background threads across the fork. */
+		wait_for_child_exit(pid);
+		expect_true(get_background_thread_enabled(),
+		    "Parent should still have background thread enabled.");
+		expect_zu_gt(n_background_threads, 0,
+		    "Parent should still have background threads after fork.");
+	}
+#else
+	test_skip("fork(2) is irrelevant to Windows");
+#endif
+}
+TEST_END
+
 int
 main(void) {
 	return test_no_reentrancy(test_fork, test_fork_child_usability,
 	    test_fork_multithreaded, test_fork_postfork_descriptor_relink,
 	    test_fork_postfork_descriptor_relink_multithreaded,
-	    test_fork_postfork_double_fork);
+	    test_fork_postfork_double_fork, test_fork_background_thread);
 }

test/unit/hpa.c

@@ -1,5 +1,6 @@
 #include "test/jemalloc_test.h"
 
+#include "jemalloc/internal/deferral.h"
 #include "jemalloc/internal/hpa.h"
 #include "jemalloc/internal/nstime.h"
 
@@ -1131,14 +1132,14 @@ TEST_BEGIN(test_deferred_until_time) {
 	/* Current time = 900ms, purge_eligible at 1300ms */
 	nstime_init(&defer_curtime, 900UL * 1000 * 1000);
 	uint64_t until_ns = hpa_time_until_deferred_work(tsdn, shard);
-	expect_u64_eq(until_ns, BACKGROUND_THREAD_DEFERRED_MIN,
+	expect_u64_eq(until_ns, DEFERRED_WORK_MIN,
 	    "First pass did not happen");
 
 	/* Fake that first pass happened more than min_purge_interval_ago */
 	nstime_init(&shard->last_purge, 350UL * 1000 * 1000);
 	shard->stats.npurge_passes = 1;
 	until_ns = hpa_time_until_deferred_work(tsdn, shard);
-	expect_u64_eq(until_ns, BACKGROUND_THREAD_DEFERRED_MIN,
+	expect_u64_eq(until_ns, DEFERRED_WORK_MIN,
 	    "No need to heck anything it is more than interval");
 
 	nstime_init(&shard->last_purge, 900UL * 1000 * 1000);

test/unit/pac.c

@@ -2,6 +2,13 @@
 
 #include "jemalloc/internal/pa.h"
 
+/*
+ * pac_decay_deferred is JET_EXTERN (test-only; see src/pac.c) and has no public
+ * header declaration, so declare it locally for the deterministic test below.
+ */
+extern void pac_decay_deferred(tsdn_t *tsdn, pac_t *pac,
+    pac_purge_eagerness_t eagerness, pac_deferred_work_result_t *result);
+
 #define PAC_TEST_ARENA_IND 124
 
 static bool pac_test_dalloc_fail;
@@ -114,12 +121,14 @@ pac_test_data_init(bool custom_hooks, ssize_t dirty_decay_ms,
 
 	nstime_t time;
 	nstime_init(&time, 0);
-	assert_false(pa_central_init(&data->central, data->base, /* hpa */ false,
+	assert_false(pa_central_init(&data->central, data->base,
-	    &hpa_hooks_default), "Unexpected pa_central_init failure");
+	    /* hpa */ false, &hpa_hooks_default),
+	    "Unexpected pa_central_init failure");
 	assert_false(pa_shard_init(tsdn, &data->shard, &data->central,
 	    &data->emap, data->base, PAC_TEST_ARENA_IND, &data->stats,
 	    &data->stats_mtx, &time, SC_LARGE_MAXCLASS + PAGE,
-	    dirty_decay_ms, muzzy_decay_ms), "Unexpected pa_shard_init failure");
+	    dirty_decay_ms, muzzy_decay_ms),
+	    "Unexpected pa_shard_init failure");
 
 	return data;
 }
@@ -382,6 +391,45 @@ TEST_BEGIN(test_pac_large_guarded_dalloc_unguards_before_caching) {
 }
 TEST_END
 
+TEST_BEGIN(test_pac_deferred_never_does_not_purge) {
+	/*
+	 * PAC_PURGE_NEVER is the eagerness the application free path selects
+	 * when a background thread is enabled, meaning current thread does not
+	 * purge.  It is only reachable through the exposed pac_decay_deferred
+	 * primitive (the production entry pac_do_deferred_work picks eagerness
+	 * itself), so this is a deliberate internal test of that critical,
+	 * stable edge.
+	 */
+	pac_test_hooks_reset();
+	/*
+	 * Finite dirty decay so a purge is gated by eagerness, not decay_ms<=0.
+	 */
+	pac_test_data_t *data = pac_test_data_init(
+	    /* custom_hooks */ true, /* dirty_decay_ms */ 100 * 1000, -1);
+	pac_t *pac = &data->shard.pac;
+
+	edata_t *edata = pac_alloc_expect(data, PAGE, /* guarded */ false);
+	size_t ndirty_before_dalloc = ecache_npages_get(&pac->ecache_dirty);
+
+	/* Dalloc caches the extent as dirty and requests deferred work. */
+	bool deferred_work_generated = false;
+	pac_dalloc(tsdn_fetch(), pac, edata, &deferred_work_generated);
+	expect_true(deferred_work_generated,
+	    "Non-pinned dalloc should request deferred work");
+	size_t ndirty_after_dalloc = ecache_npages_get(&pac->ecache_dirty);
+	expect_zu_gt(ndirty_after_dalloc, ndirty_before_dalloc,
+	    "Dalloc should add dirty pages");
+
+	/* PAC_PURGE_NEVER must leave the dirty pages untouched. */
+	pac_deferred_work_result_t result;
+	pac_decay_deferred(tsdn_fetch(), pac, PAC_PURGE_NEVER, &result);
+	expect_zu_eq(ecache_npages_get(&pac->ecache_dirty), ndirty_after_dalloc,
+	    "PAC_PURGE_NEVER must not purge any dirty pages");
+
+	pac_test_data_destroy(data);
+}
+TEST_END
+
 int
 main(void) {
 	return test_no_reentrancy(test_pac_dirty_muzzy_alloc_priority,
@@ -391,5 +439,6 @@ main(void) {
 	    test_pac_decay_retains_when_dalloc_fails,
 	    test_pac_non_pinned_dalloc_signals_deferred_work,
 	    test_pac_non_pinned_shrink_signals_deferred_work,
-	    test_pac_large_guarded_dalloc_unguards_before_caching);
+	    test_pac_large_guarded_dalloc_unguards_before_caching,
+	    test_pac_deferred_never_does_not_purge);
 }

test/unit/pac_decay.c

@@ -1,6 +1,8 @@
 #include "test/jemalloc_test.h"
 #include "test/arena_util.h"
 
+#include "jemalloc/internal/deferral.h"
+#include "jemalloc/internal/pac.h"
 #include "jemalloc/internal/ticker.h"
 
 static nstime_monotonic_t *nstime_monotonic_orig;
@@ -439,8 +441,35 @@ TEST_BEGIN(test_decay_never) {
 }
 TEST_END
 
+/*
+ * pac_time_until_deferred_work is the public PAC scheduling entry the
+ * background thread uses to decide its next wakeup.  Exercised here on a real
+ * arena through the public interface (no direct decay-state manipulation).
+ */
+TEST_BEGIN(test_pac_time_until_deferred_work) {
+	test_skip_if(is_background_thread_enabled());
+	test_skip_if(opt_hpa);
+
+	unsigned arena_ind = do_arena_create(1000, -1);
+	tsdn_t  *tsdn = tsd_tsdn(tsd_fetch());
+	pac_t   *pac = &arena_get(tsdn, arena_ind, false)->pa_shard.pac;
+
+	/* Idle: nothing pending -> the scheduler reports the maximum wait. */
+	expect_u64_eq(pac_time_until_deferred_work(tsdn, pac), DEFERRED_WORK_MAX,
+	    "Idle PAC should defer for the maximum interval");
+
+	/* Dirty pages under a finite decay_ms -> a finite wait is scheduled. */
+	generate_dirty(arena_ind, PAGE);
+	expect_u64_lt(pac_time_until_deferred_work(tsdn, pac), DEFERRED_WORK_MAX,
+	    "Pending dirty decay should schedule a finite deferred-work time");
+
+	do_arena_destroy(arena_ind);
+}
+TEST_END
+
 int
 main(void) {
 	return test(test_decay_ticks, test_decay_ticker,
-	    test_decay_nonmonotonic, test_decay_now, test_decay_never);
+	    test_decay_nonmonotonic, test_decay_now, test_decay_never,
+	    test_pac_time_until_deferred_work);
 }

test/unit/pac_decay.sh

@@ -1,3 +1,3 @@
 #!/bin/sh
 
-export MALLOC_CONF="dirty_decay_ms:1000,muzzy_decay_ms:1000,tcache_max:1024"
+export MALLOC_CONF="dirty_decay_ms:1000,muzzy_decay_ms:1000,tcache_max:1024,background_thread:false"