Simplify import flushing via subinterpreter replacement

Replace del_import with generation-based pool slot recycling.
When flush_imports is called, all pool slots are marked stale.
When a stale slot's usage count drops to 0, the subinterpreter
is destroyed and reinitialized with a fresh one.

Changes:
- Add usage_count, marked_stale, generation to subinterp_slot_t
- Add subinterp_pool_acquire/release/flush_generation functions
- Remove del_import/1,2 from py.erl (no longer needed)
- Update flush_imports to use generation-based approach
- Remove del_import tests from test suite

Author: benoitc

c_src/py_nif.c

@@ -441,6 +441,9 @@ static void context_destructor(ErlNifEnv *env, void *obj) {
         ctx->globals = NULL;
         ctx->locals = NULL;
 
+        /* Release usage count - if slot is stale and count drops to 0,
+         * the subinterpreter will be destroyed automatically */
+        subinterp_pool_release(ctx->pool_slot);
         subinterp_pool_free(ctx->pool_slot);
         ctx->pool_slot = -1;
         ctx->destroyed = true;
@@ -4157,9 +4160,13 @@ static ERL_NIF_TERM nif_context_create(ErlNifEnv *env, int argc, const ERL_NIF_T
 
         ctx->pool_slot = slot;
 
+        /* Acquire usage count for the pool slot */
+        subinterp_pool_acquire(slot);
+
         /* Get the pool slot for interpreter access */
         subinterp_slot_t *pool_slot = subinterp_pool_get(slot);
         if (pool_slot == NULL || !pool_slot->initialized) {
+            subinterp_pool_release(slot);  /* Release usage count */
             subinterp_pool_free(slot);
             close(ctx->callback_pipe[0]);
             close(ctx->callback_pipe[1]);
@@ -4183,6 +4190,7 @@ static ERL_NIF_TERM nif_context_create(ErlNifEnv *env, int argc, const ERL_NIF_T
             Py_XDECREF(ctx->module_cache);
             PyThreadState_Swap(saved);
             PyGILState_Release(gstate);
+            subinterp_pool_release(slot);  /* Release usage count */
             subinterp_pool_free(slot);
             close(ctx->callback_pipe[0]);
             close(ctx->callback_pipe[1]);
@@ -4313,7 +4321,11 @@ static ERL_NIF_TERM nif_context_destroy(ErlNifEnv *env, int argc, const ERL_NIF_
         ctx->locals = NULL;
         ctx->module_cache = NULL;
 
-        /* Release the pool slot back to the pool */
+        /* Release the pool slot back to the pool.
+         * subinterp_pool_release decrements usage count and may destroy
+         * the subinterpreter if it's marked stale and count drops to 0.
+         * subinterp_pool_free marks the slot as available for new contexts. */
+        subinterp_pool_release(ctx->pool_slot);
         subinterp_pool_free(ctx->pool_slot);
         ctx->pool_slot = -1;
 
@@ -5074,6 +5086,31 @@ static ERL_NIF_TERM nif_interp_flush_imports(ErlNifEnv *env, int argc, const ERL
     return ATOM_OK;
 }
 
+/**
+ * @brief Flush import generation and mark all pool slots stale
+ *
+ * nif_subinterp_pool_flush_generation() -> {ok, NewGeneration}
+ *
+ * Increments the global import generation counter and marks all initialized
+ * pool slots as stale. When a stale slot's usage count drops to 0, the
+ * subinterpreter is automatically destroyed and the slot becomes available
+ * for a fresh subinterpreter.
+ *
+ * This enables flush_imports to work by replacing subinterpreters rather
+ * than trying to manipulate sys.modules directly (which can crash C extensions).
+ */
+static ERL_NIF_TERM nif_subinterp_pool_flush_generation(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]) {
+    (void)argc;
+    (void)argv;
+
+#ifdef HAVE_SUBINTERPRETERS
+    uint64_t new_gen = subinterp_pool_flush_generation();
+    return enif_make_tuple2(env, ATOM_OK, enif_make_uint64(env, new_gen));
+#else
+    return enif_make_tuple2(env, ATOM_OK, enif_make_uint64(env, 0));
+#endif
+}
+
 /**
  * @brief Execute Python statements using a process-local environment
  *
@@ -7205,6 +7242,7 @@ static ErlNifFunc nif_funcs[] = {
     {"create_local_env", 1, nif_create_local_env, 0},
     {"interp_apply_imports", 2, nif_interp_apply_imports, ERL_NIF_DIRTY_JOB_CPU_BOUND},
     {"interp_flush_imports", 2, nif_interp_flush_imports, ERL_NIF_DIRTY_JOB_CPU_BOUND},
+    {"subinterp_pool_flush_generation", 0, nif_subinterp_pool_flush_generation, 0},
     {"context_call_method", 4, nif_context_call_method, ERL_NIF_DIRTY_JOB_CPU_BOUND},
     {"context_to_term", 1, nif_context_to_term, 0},
     {"context_interp_id", 1, nif_context_interp_id, 0},

c_src/py_subinterp_pool.c

@@ -47,6 +47,9 @@ static int g_pool_size = 0;
 /** @brief Pool initialized flag */
 static _Atomic bool g_pool_initialized = false;
 
+/** @brief Global import generation counter */
+static _Atomic uint64_t g_import_generation = 0;
+
 /* ============================================================================
  * Forward declarations for functions defined in other modules
  * ============================================================================ */
@@ -196,6 +199,9 @@ int subinterp_pool_init(int size) {
         }
 
         slot->initialized = true;
+        atomic_store(&slot->usage_count, 0);
+        atomic_store(&slot->marked_stale, false);
+        slot->generation = atomic_load(&g_import_generation);
 
         /* Swap back to main thread state before creating next subinterpreter */
         PyThreadState_Swap(main_tstate);
@@ -345,4 +351,193 @@ bool subinterp_pool_is_initialized(void) {
     return atomic_load(&g_pool_initialized);
 }
 
+void subinterp_pool_acquire(int slot) {
+    if (slot < 0 || slot >= g_pool_size) {
+        return;
+    }
+
+    subinterp_slot_t *s = &g_subinterp_pool[slot];
+    if (!s->initialized) {
+        return;
+    }
+
+    atomic_fetch_add(&s->usage_count, 1);
+}
+
+/**
+ * @brief Reinitialize a pool slot with a fresh subinterpreter
+ *
+ * Called after destroying a stale subinterpreter to create a fresh one.
+ * Assumes GIL is already held.
+ *
+ * @param slot_idx Slot index
+ * @return 0 on success, -1 on failure
+ */
+static int subinterp_pool_reinit_slot(int slot_idx) {
+    subinterp_slot_t *slot = &g_subinterp_pool[slot_idx];
+
+    /* Save main thread state */
+    PyThreadState *main_tstate = PyThreadState_Get();
+
+    /* Configure subinterpreter with SHARED GIL */
+    PyInterpreterConfig config = {
+        .use_main_obmalloc = 0,
+        .allow_fork = 0,
+        .allow_exec = 0,
+        .allow_threads = 1,
+        .allow_daemon_threads = 0,
+        .check_multi_interp_extensions = 1,
+        .gil = PyInterpreterConfig_SHARED_GIL,
+    };
+
+    PyThreadState *tstate = NULL;
+    PyStatus status = Py_NewInterpreterFromConfig(&tstate, &config);
+
+    if (PyStatus_Exception(status) || tstate == NULL) {
+        fprintf(stderr, "subinterp_pool_reinit_slot: failed to create subinterp for slot %d\n", slot_idx);
+        PyThreadState_Swap(main_tstate);
+        return -1;
+    }
+
+    /* tstate is now the current thread state for the new interpreter */
+    slot->interp = PyThreadState_GetInterpreter(tstate);
+    slot->tstate = tstate;
+
+    /* Initialize globals/locals */
+    slot->globals = PyDict_New();
+    slot->locals = PyDict_New();
+    slot->module_cache = PyDict_New();
+
+    if (slot->globals == NULL || slot->locals == NULL || slot->module_cache == NULL) {
+        Py_XDECREF(slot->module_cache);
+        Py_XDECREF(slot->globals);
+        Py_XDECREF(slot->locals);
+        Py_EndInterpreter(tstate);
+        PyThreadState_Swap(main_tstate);
+        return -1;
+    }
+
+    /* Import __builtins__ into globals */
+    PyObject *builtins = PyEval_GetBuiltins();
+    PyDict_SetItemString(slot->globals, "__builtins__", builtins);
+
+    /* Create erlang module in this subinterpreter */
+    if (create_erlang_module() >= 0) {
+        /* Register ReactorBuffer and PyBuffer */
+        ReactorBuffer_register_with_reactor();
+        PyBuffer_register_with_module();
+
+        /* Import erlang module into globals */
+        PyObject *erlang_module = PyImport_ImportModule("erlang");
+        if (erlang_module != NULL) {
+            PyDict_SetItemString(slot->globals, "erlang", erlang_module);
+            Py_DECREF(erlang_module);
+        }
+    }
+
+    /* Initialize event loop for this subinterpreter */
+    init_subinterpreter_event_loop(NULL);
+
+    slot->initialized = true;
+    atomic_store(&slot->usage_count, 0);
+    atomic_store(&slot->marked_stale, false);
+    slot->generation = atomic_load(&g_import_generation);
+
+    /* Swap back to main thread state */
+    PyThreadState_Swap(main_tstate);
+
+#ifdef DEBUG
+    fprintf(stderr, "subinterp_pool_reinit_slot: reinitialized slot %d\n", slot_idx);
+#endif
+
+    return 0;
+}
+
+void subinterp_pool_release(int slot) {
+    if (slot < 0 || slot >= g_pool_size) {
+        return;
+    }
+
+    subinterp_slot_t *s = &g_subinterp_pool[slot];
+    if (!s->initialized) {
+        return;
+    }
+
+    int prev_count = atomic_fetch_sub(&s->usage_count, 1);
+
+    /* If usage drops to 0 and slot is marked stale, destroy and reinitialize
+     * the subinterpreter. We need the GIL to safely do Python operations. */
+    if (prev_count == 1 && atomic_load(&s->marked_stale)) {
+        /* Acquire the GIL */
+        PyGILState_STATE gstate = PyGILState_Ensure();
+
+        /* Save current thread state and swap to the subinterpreter */
+        PyThreadState *saved = PyThreadState_Swap(s->tstate);
+
+        /* Clean up Python objects */
+        Py_XDECREF(s->module_cache);
+        Py_XDECREF(s->globals);
+        Py_XDECREF(s->locals);
+
+        /* End the interpreter - this frees s->tstate */
+        Py_EndInterpreter(s->tstate);
+
+        /* Clear slot state */
+        s->interp = NULL;
+        s->tstate = NULL;
+        s->globals = NULL;
+        s->locals = NULL;
+        s->module_cache = NULL;
+        s->initialized = false;
+        atomic_store(&s->marked_stale, false);
+
+        /* Swap back to saved thread state (may be NULL if we didn't have one) */
+        PyThreadState_Swap(saved);
+
+        /* Reinitialize the slot with a fresh subinterpreter */
+        if (subinterp_pool_reinit_slot(slot) < 0) {
+            /* Reinit failed - clear allocation bit so slot is skipped */
+            uint64_t mask = ~(1ULL << slot);
+            atomic_fetch_and(&g_pool_allocation, mask);
+            fprintf(stderr, "subinterp_pool_release: failed to reinit slot %d\n", slot);
+        }
+        /* If reinit succeeded, slot stays allocated but now has fresh interp */
+
+        /* Release the GIL */
+        PyGILState_Release(gstate);
+
+#ifdef DEBUG
+        fprintf(stderr, "subinterp_pool_release: recycled stale slot %d\n", slot);
+#endif
+    }
+}
+
+uint64_t subinterp_pool_flush_generation(void) {
+    if (!atomic_load(&g_pool_initialized)) {
+        return 0;
+    }
+
+    /* Increment generation */
+    uint64_t new_gen = atomic_fetch_add(&g_import_generation, 1) + 1;
+
+    /* Mark all initialized slots as stale */
+    for (int i = 0; i < g_pool_size; i++) {
+        subinterp_slot_t *slot = &g_subinterp_pool[i];
+        if (slot->initialized) {
+            atomic_store(&slot->marked_stale, true);
+        }
+    }
+
+#ifdef DEBUG
+    fprintf(stderr, "subinterp_pool_flush_generation: incremented to %llu, marked all slots stale\n",
+            (unsigned long long)new_gen);
+#endif
+
+    return new_gen;
+}
+
+uint64_t subinterp_pool_get_generation(void) {
+    return atomic_load(&g_import_generation);
+}
+
 #endif /* HAVE_SUBINTERPRETERS */

c_src/py_subinterp_pool.h

@@ -84,6 +84,15 @@ typedef struct {
 
     /** @brief Whether this slot is initialized and ready for use */
     bool initialized;
+
+    /** @brief Number of contexts currently using this slot */
+    _Atomic int usage_count;
+
+    /** @brief Marked for destruction when usage_count reaches 0 */
+    _Atomic bool marked_stale;
+
+    /** @brief Import generation when slot was created */
+    uint64_t generation;
 } subinterp_slot_t;
 
 /**
@@ -163,6 +172,43 @@ void subinterp_pool_shutdown(void);
  */
 bool subinterp_pool_is_initialized(void);
 
+/**
+ * @brief Increment usage count for a slot
+ *
+ * Called when a context starts using a pool slot.
+ *
+ * @param slot Slot index
+ */
+void subinterp_pool_acquire(int slot);
+
+/**
+ * @brief Decrement usage count for a slot
+ *
+ * Called when a context stops using a pool slot.
+ * If the slot is marked stale and usage_count reaches 0,
+ * the subinterpreter is destroyed and the slot is cleared.
+ *
+ * @param slot Slot index
+ */
+void subinterp_pool_release(int slot);
+
+/**
+ * @brief Increment import generation and mark all pool slots stale
+ *
+ * Called by flush_imports to trigger subinterpreter replacement.
+ * Stale slots will be destroyed when their usage_count reaches 0.
+ *
+ * @return The new generation number
+ */
+uint64_t subinterp_pool_flush_generation(void);
+
+/**
+ * @brief Get the current import generation
+ *
+ * @return Current generation number
+ */
+uint64_t subinterp_pool_get_generation(void);
+
 #endif /* HAVE_SUBINTERPRETERS */
 
 #endif /* PY_SUBINTERP_POOL_H */