Symmetrize deferred-work execution by moving it from arena to PAC

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);
@@ -188,15 +182,14 @@ bool    pa_decay_ms_set(tsdn_t *tsdn, pa_shard_t *shard, extent_state_t state,
 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.
+ * 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);
-void     pa_shard_do_deferred_work(tsdn_t *tsdn, pa_shard_t *shard);
+void     pa_shard_do_deferred_work(
+    tsdn_t *tsdn, pa_shard_t *shard, pac_purge_eagerness_t eagerness);
 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);
 

include/jemalloc/internal/pac.h

@@ -231,6 +231,32 @@ 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_do_deferred_work() pass.  Reported per decay state so the
+ * caller (on the user-inline path) 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;
+};
+
+/*
+ * Non-forced deferred decay work for both the dirty and muzzy states.
+ * Corresponding decay->mtx are acquired internally.
+ */
+void pac_do_deferred_work(tsdn_t *tsdn, pac_t *pac,
+    pac_purge_eagerness_t eagerness, pac_deferred_work_result_t *result);
+
+/*
+ * 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).
@@ -247,6 +273,13 @@ 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 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;
+}
+
 void pac_destroy(tsdn_t *tsdn, pac_t *pac);
 
 void pac_sec_flush(tsdn_t *tsdn, pac_t *pac);

src/arena.c

@@ -83,8 +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);
 
@@ -339,7 +337,7 @@ arena_handle_deferred_work(tsdn_t *tsdn, arena_t *arena) {
 	    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);
+		pac_decay_all_now(tsdn, &arena->pa_shard.pac, extent_state_dirty);
 	}
 	arena_background_thread_inactivity_check(tsdn, arena, false);
 }
@@ -506,74 +504,67 @@ 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).
+		 * 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);
-	}
-	if (arena_decay_dirty(tsdn, arena, is_background_thread, all)) {
+		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;
 	}
-	arena_decay_muzzy(tsdn, arena, is_background_thread, all);
+
+	/*
+	 * Deferred (non-forced) decay-purge.  The PAC layer owns the decay
+	 * orchestration (lock acquisition, dirty-before-muzzy ordering, the
+	 * 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(
+	    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
@@ -648,8 +639,13 @@ label_done:
 /* 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);
+	/*
+	 * The background thread forces decay (PAC_PURGE_ALWAYS) and drives both
+	 * PAC and HPA deferred work through the symmetric pa_shard facade.  No
+	 * PAC result is needed here to notify background thread for an early
+	 * wake because this function should be called in background thread.
+	 */
+	pa_shard_do_deferred_work(tsdn, &arena->pa_shard, PAC_PURGE_ALWAYS);
 }
 
 static void

src/pa.c

@@ -250,7 +250,15 @@ pa_shard_set_deferral_allowed(
 }
 
 void
-pa_shard_do_deferred_work(tsdn_t *tsdn, pa_shard_t *shard) {
+pa_shard_do_deferred_work(
+    tsdn_t *tsdn, pa_shard_t *shard, pac_purge_eagerness_t eagerness) {
+	/*
+	 * The PAC result is only consumed on the application notification path,
+	 * which calls pac_do_deferred_work directly.  This facade just drives the
+	 * work, so the result stays local.
+	 */
+	pac_deferred_work_result_t result;
+	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

@@ -779,6 +779,68 @@ 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;
+}
+
+void
+pac_do_deferred_work(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);
+}
+
+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);
+}
+
 bool
 pac_decay_ms_set(tsdn_t *tsdn, pac_t *pac, extent_state_t state,
     ssize_t decay_ms, pac_purge_eagerness_t eagerness) {