diff --git a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java index c99bf16..6614177 100644 --- a/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java +++ b/flink-processor/src/main/java/org/mobilitydb/flink/meos/wirings/MeosWiringRuntime.java @@ -26,37 +26,72 @@ package org.mobilitydb.flink.meos.wirings; import functions.GeneratedFunctions; +import functions.error_handler_fn; /** - * Per-thread MEOS initialization for the {@code org.mobilitydb.flink.meos.wirings} - * operators. + * MEOS initialization for the {@code org.mobilitydb.flink.meos.wirings} operators. * - *

MEOS keeps its timezone / session state per OS thread. Each Flink - * subtask runs on its own task thread, so every wiring operator must - * initialize MEOS on that thread from its {@code open()} — the JVM-wide - * probe in {@code MeosOpsRuntime} only covers the thread that first - * touches a facade class (typically the job's main thread), not the task - * threads where the operators actually run. + *

MEOS setup has two lifetimes: process-global state (the allocator and the + * error handler) is installed once per JVM, while thread-local state (the + * timezone and collation caches; the PROJ, GEOS and GSL contexts) belongs to + * each task thread. Each Flink subtask runs on its own task thread, so every + * wiring operator calls {@link #ensureInitializedOnThread()} from its + * {@code open()} to install the thread-local part on that thread. * - *

{@link #ensureInitializedOnThread()} is idempotent per thread (guarded - * by a {@link ThreadLocal}), so it is safe to call from every operator's - * {@code open()} even when operators are chained onto the same thread. It - * installs a no-op error handler so a MEOS-side error surfaces as a thrown - * exception rather than terminating the JVM. + *

{@link #ensureInitializedOnThread()} is idempotent per thread (guarded by a + * {@link ThreadLocal}), so it is safe to call from every operator's + * {@code open()} even when operators are chained onto the same thread. The + * process-global no-exit error handler makes a MEOS-side error surface as a + * thrown exception rather than terminating the JVM. */ public final class MeosWiringRuntime { - private static final ThreadLocal INITIALIZED = - ThreadLocal.withInitial(() -> Boolean.FALSE); + /** + * No-exit MEOS error handler. MEOS's default handler calls exit(EXIT_FAILURE) + * on an ERROR, which would tear the JVM down if a MEOS error fired inside a + * task thread; this handler returns instead, and the error still surfaces + * because MEOS sets meos_errno. Held as a static field so JNR keeps the + * native callback alive for the process lifetime. + */ + private static final error_handler_fn NOEXIT_ERROR_HANDLER = + (level, code, message) -> { /* do not exit the JVM */ }; + + /** + * Process-global MEOS setup, installed exactly once per JVM: the allocator + * and the error handler are process-global, not thread-local. The holder's + * class initializer runs under the JVM class-initialization lock. meos_initialize() + * runs first — it installs MEOS's exiting default handler — and the no-exit + * handler then replaces it, so a thread reaching its per-thread setup always + * sees the no-exit handler and the exiting default is never observable. + */ + private static final class ProcessInit { + static { + GeneratedFunctions.meos_initialize(); + GeneratedFunctions.meos_initialize_error_handler(NOEXIT_ERROR_HANDLER); + } + /** Invoking this forces the class initializer above to run once. */ + static void ensure() { /* side effect: class initialization */ } + } + + /** + * Per-thread MEOS setup, run once per task thread: only the thread-local + * caches. The timezone and collation caches are thread-local and set + * explicitly per thread; the PROJ, GEOS and GSL contexts are thread-local too + * and created lazily by MEOS on first use. Full meos_initialize() is NOT run + * per thread — it re-installs the exiting default error handler that every + * other thread relies on being the no-exit one. + */ + private static final ThreadLocal INITIALIZED = ThreadLocal.withInitial(() -> { + ProcessInit.ensure(); + GeneratedFunctions.meos_initialize_timezone("UTC"); + GeneratedFunctions.meos_initialize_collation(); + return Boolean.TRUE; + }); private MeosWiringRuntime() { /* utility */ } /** Initialize MEOS on the calling thread exactly once. */ public static void ensureInitializedOnThread() { - if (!INITIALIZED.get()) { - GeneratedFunctions.meos_initialize_error_handler((level, code, message) -> { }); - GeneratedFunctions.meos_initialize(); - INITIALIZED.set(Boolean.TRUE); - } + INITIALIZED.get(); } }