Add event loop performance optimizations

- Fix run_until_complete callback removal bug (use single reference)
- Cache ast.literal_eval lookup at module initialization
- Add O(1) timer cancellation via handle-to-callback_id reverse map
- Detach pending queue under lock, build Erlang terms outside lock
- Replace O(n) duplicate scan with hash set (128-slot open addressing)
- Add PERF_BUILD cmake option for aggressive optimizations (-O3, LTO)

Author: benoitc

c_src/CMakeLists.txt

@@ -47,6 +47,17 @@ if(NOT CMAKE_BUILD_TYPE)
     set(CMAKE_BUILD_TYPE Release)
 endif()
 
+# Performance build option (for maximum optimization)
+option(PERF_BUILD "Enable aggressive performance optimizations (-O3, LTO, native arch)" OFF)
+
+if(PERF_BUILD)
+    message(STATUS "Performance build enabled - using aggressive optimizations")
+    # Override compiler flags for maximum performance
+    set(CMAKE_C_FLAGS_RELEASE "-O3 -DNDEBUG")
+    # Enable Link-Time Optimization
+    set(CMAKE_INTERPROCEDURAL_OPTIMIZATION TRUE)
+endif()
+
 # Find Erlang
 include(FindErlang)
 include_directories(${ERLANG_ERTS_INCLUDE_PATH})
@@ -99,11 +110,24 @@ target_include_directories(py_nif PRIVATE
 )
 
 # Compiler flags
-target_compile_options(py_nif PRIVATE
-    -O2
-    -Wall
-    -fPIC
-)
+if(PERF_BUILD)
+    # Performance build: aggressive optimizations
+    target_compile_options(py_nif PRIVATE
+        -O3
+        -Wall
+        -fPIC
+        -march=native
+        -ffast-math
+        -funroll-loops
+    )
+else()
+    # Standard build
+    target_compile_options(py_nif PRIVATE
+        -O2
+        -Wall
+        -fPIC
+    )
+endif()
 
 # Platform-specific settings
 if(APPLE)

c_src/py_callback.c

@@ -83,6 +83,44 @@
  * @note This file is included from py_nif.c (single compilation unit)
  */
 
+/* ============================================================================
+ * Cached Python Function References
+ *
+ * Cache frequently-used Python functions to avoid repeated module import
+ * and attribute lookup overhead on every callback.
+ * ============================================================================ */
+
+/** @brief Cached reference to ast.literal_eval function */
+static PyObject *g_ast_literal_eval = NULL;
+
+/**
+ * @brief Initialize cached Python function references
+ *
+ * Called during module initialization. Must be called with GIL held.
+ */
+static void init_callback_cache(void) {
+    if (g_ast_literal_eval == NULL) {
+        PyObject *ast_mod = PyImport_ImportModule("ast");
+        if (ast_mod != NULL) {
+            g_ast_literal_eval = PyObject_GetAttrString(ast_mod, "literal_eval");
+            Py_DECREF(ast_mod);
+        }
+        if (g_ast_literal_eval == NULL) {
+            PyErr_Clear();  /* Non-fatal if unavailable */
+        }
+    }
+}
+
+/**
+ * @brief Cleanup cached Python function references
+ *
+ * Called during module cleanup. Must be called with GIL held.
+ */
+static void cleanup_callback_cache(void) {
+    Py_XDECREF(g_ast_literal_eval);
+    g_ast_literal_eval = NULL;
+}
+
 /* ============================================================================
  * Callback Name Registry
  *
@@ -580,24 +618,18 @@ static PyObject *parse_callback_response(unsigned char *response_data, size_t re
 
     PyObject *result = NULL;
     if (status == 0) {
-        /* Try to evaluate the result string as Python literal */
-        PyObject *ast_module = PyImport_ImportModule("ast");
-        if (ast_module != NULL) {
-            PyObject *literal_eval = PyObject_GetAttrString(ast_module, "literal_eval");
-            if (literal_eval != NULL) {
-                PyObject *arg = PyUnicode_FromStringAndSize(result_str, result_len);
-                if (arg != NULL) {
-                    result = PyObject_CallFunctionObjArgs(literal_eval, arg, NULL);
-                    Py_DECREF(arg);
-                    if (result == NULL) {
-                        /* If literal_eval fails, return as string */
-                        PyErr_Clear();
-                        result = PyUnicode_FromStringAndSize(result_str, result_len);
-                    }
+        /* Try to evaluate the result string as Python literal using cached function */
+        if (g_ast_literal_eval != NULL) {
+            PyObject *arg = PyUnicode_FromStringAndSize(result_str, result_len);
+            if (arg != NULL) {
+                result = PyObject_CallFunctionObjArgs(g_ast_literal_eval, arg, NULL);
+                Py_DECREF(arg);
+                if (result == NULL) {
+                    /* If literal_eval fails, return as string */
+                    PyErr_Clear();
+                    result = PyUnicode_FromStringAndSize(result_str, result_len);
                 }
-                Py_DECREF(literal_eval);
             }
-            Py_DECREF(ast_module);
         }
         if (result == NULL) {
             result = PyUnicode_FromStringAndSize(result_str, result_len);
@@ -1291,6 +1323,9 @@ static struct PyModuleDef ErlangModuleDef = {
  * Called during Python initialization.
  */
 static int create_erlang_module(void) {
+    /* Initialize cached Python function references */
+    init_callback_cache();
+
     /* Initialize ErlangFunction type */
     if (PyType_Ready(&ErlangFunctionType) < 0) {
         return -1;

c_src/py_event_loop.c

@@ -785,6 +785,9 @@ ERL_NIF_TERM nif_poll_events(ErlNifEnv *env, int argc,
     return enif_make_tuple2(env, ATOM_OK, enif_make_int(env, num_events));
 }
 
+/* Forward declaration for hash set clear function (defined with other hash functions) */
+static inline void pending_hash_clear(erlang_event_loop_t *loop);
+
 /**
  * get_pending(LoopRef) -> [{CallbackId, Type}]
  *
@@ -800,10 +803,28 @@ ERL_NIF_TERM nif_get_pending(ErlNifEnv *env, int argc,
         return enif_make_list(env, 0);
     }
 
+    /*
+     * Phase 1: Detach pending list under lock (fast - just pointer swap)
+     * This minimizes lock contention by doing minimal work under the mutex.
+     */
     pthread_mutex_lock(&loop->mutex);
 
+    pending_event_t *snapshot_head = loop->pending_head;
+    loop->pending_head = NULL;
+    loop->pending_tail = NULL;
+    atomic_store(&loop->pending_count, 0);
+
+    /* Clear the hash set since we're consuming all pending events */
+    pending_hash_clear(loop);
+
+    pthread_mutex_unlock(&loop->mutex);
+
+    /*
+     * Phase 2: Build Erlang list outside lock (no contention)
+     * Term creation and memory operations happen without holding the mutex.
+     */
     ERL_NIF_TERM list = enif_make_list(env, 0);
-    pending_event_t *current = loop->pending_head;
+    pending_event_t *current = snapshot_head;
 
     while (current != NULL) {
         ERL_NIF_TERM type_atom;
@@ -833,12 +854,6 @@ ERL_NIF_TERM nif_get_pending(ErlNifEnv *env, int argc,
         current = next;
     }
 
-    loop->pending_head = NULL;
-    loop->pending_tail = NULL;
-    atomic_store(&loop->pending_count, 0);
-
-    pthread_mutex_unlock(&loop->mutex);
-
     /* Reverse the list to maintain order */
     ERL_NIF_TERM reversed = enif_make_list(env, 0);
     ERL_NIF_TERM head;
@@ -1033,19 +1048,114 @@ static inline void return_pending_event(erlang_event_loop_t *loop,
     }
 }
 
+/* ============================================================================
+ * Pending Event Hash Set (O(1) duplicate detection)
+ *
+ * Uses open addressing with linear probing. Key is (callback_id, type)
+ * combined into a single uint64_t.
+ * ============================================================================ */
+
+/**
+ * @brief Compute hash key from callback_id and event type
+ */
+static inline uint64_t pending_hash_key(uint64_t callback_id, event_type_t type) {
+    /* Combine callback_id and type into a single key */
+    return (callback_id << 2) | (uint64_t)type;
+}
+
+/**
+ * @brief Compute hash bucket index
+ */
+static inline uint32_t pending_hash_index(uint64_t key) {
+    /* Simple hash: XOR fold and modulo */
+    return (uint32_t)((key ^ (key >> 32)) % PENDING_HASH_SIZE);
+}
+
+/**
+ * @brief Check if a (callback_id, type) pair exists in the hash set
+ *
+ * @param loop Event loop containing the hash set
+ * @param callback_id Callback ID to check
+ * @param type Event type to check
+ * @return true if exists, false otherwise
+ */
+static inline bool pending_hash_contains(erlang_event_loop_t *loop,
+                                          uint64_t callback_id, event_type_t type) {
+    if (loop->pending_hash_count == 0) {
+        return false;
+    }
+
+    uint64_t key = pending_hash_key(callback_id, type);
+    uint32_t idx = pending_hash_index(key);
+
+    /* Linear probing */
+    for (int i = 0; i < PENDING_HASH_SIZE; i++) {
+        uint32_t probe = (idx + i) % PENDING_HASH_SIZE;
+        if (!loop->pending_hash_occupied[probe]) {
+            return false;  /* Empty slot means key not present */
+        }
+        if (loop->pending_hash_keys[probe] == key) {
+            return true;
+        }
+    }
+    return false;  /* Table full, key not found */
+}
+
+/**
+ * @brief Insert a (callback_id, type) pair into the hash set
+ *
+ * @param loop Event loop containing the hash set
+ * @param callback_id Callback ID to insert
+ * @param type Event type to insert
+ * @return true if inserted, false if already exists or table full
+ */
+static inline bool pending_hash_insert(erlang_event_loop_t *loop,
+                                        uint64_t callback_id, event_type_t type) {
+    /* Don't insert if table is too full (load factor > 0.75) */
+    if (loop->pending_hash_count >= (PENDING_HASH_SIZE * 3) / 4) {
+        return false;
+    }
+
+    uint64_t key = pending_hash_key(callback_id, type);
+    uint32_t idx = pending_hash_index(key);
+
+    /* Linear probing */
+    for (int i = 0; i < PENDING_HASH_SIZE; i++) {
+        uint32_t probe = (idx + i) % PENDING_HASH_SIZE;
+        if (!loop->pending_hash_occupied[probe]) {
+            loop->pending_hash_keys[probe] = key;
+            loop->pending_hash_occupied[probe] = true;
+            loop->pending_hash_count++;
+            return true;
+        }
+        if (loop->pending_hash_keys[probe] == key) {
+            return false;  /* Already exists */
+        }
+    }
+    return false;  /* Table full */
+}
+
+/**
+ * @brief Clear the pending hash set
+ *
+ * @param loop Event loop containing the hash set
+ */
+static inline void pending_hash_clear(erlang_event_loop_t *loop) {
+    if (loop->pending_hash_count > 0) {
+        memset(loop->pending_hash_occupied, 0, sizeof(loop->pending_hash_occupied));
+        loop->pending_hash_count = 0;
+    }
+}
+
 void event_loop_add_pending(erlang_event_loop_t *loop, event_type_t type,
                             uint64_t callback_id, int fd) {
     pthread_mutex_lock(&loop->mutex);
 
-    /* Check for duplicate - don't add if same callback_id already pending */
-    pending_event_t *current = loop->pending_head;
-    while (current != NULL) {
-        if (current->callback_id == callback_id && current->type == type) {
-            /* Already have this event pending, skip */
-            pthread_mutex_unlock(&loop->mutex);
-            return;
-        }
-        current = current->next;
+    /* O(1) duplicate check using hash set */
+    if (pending_hash_contains(loop, callback_id, type)) {
+        /* Already have this event pending, skip */
+        pthread_mutex_unlock(&loop->mutex);
+        return;
     }
 
     /* Get event from freelist or allocate new (Phase 7 optimization) */
@@ -1068,6 +1178,9 @@ void event_loop_add_pending(erlang_event_loop_t *loop, event_type_t type,
         loop->pending_tail = event;
     }
 
+    /* Add to hash set for future O(1) duplicate checks */
+    pending_hash_insert(loop, callback_id, type);
+
     atomic_fetch_add(&loop->pending_count, 1);
     pthread_cond_signal(&loop->event_cond);
 
@@ -1089,6 +1202,9 @@ void event_loop_clear_pending(erlang_event_loop_t *loop) {
     loop->pending_tail = NULL;
     atomic_store(&loop->pending_count, 0);
 
+    /* Clear the hash set since all pending events are cleared */
+    pending_hash_clear(loop);
+
     pthread_mutex_unlock(&loop->mutex);
 }
 
@@ -2084,6 +2200,9 @@ static PyObject *py_get_pending(PyObject *self, PyObject *args) {
     loop->pending_tail = NULL;
     atomic_store(&loop->pending_count, 0);
 
+    /* Clear the hash set since we're consuming all pending events */
+    pending_hash_clear(loop);
+
     pthread_mutex_unlock(&loop->mutex);
 
     return list;
@@ -2455,6 +2574,7 @@ static PyObject *py_run_once(PyObject *self, PyObject *args) {
             loop->pending_head = NULL;
             loop->pending_tail = NULL;
             atomic_store(&loop->pending_count, 0);
+            pending_hash_clear(loop);
             pthread_mutex_unlock(&loop->mutex);
             return NULL;
         }
@@ -2477,6 +2597,9 @@ static PyObject *py_run_once(PyObject *self, PyObject *args) {
     loop->pending_tail = NULL;
     atomic_store(&loop->pending_count, 0);
 
+    /* Clear the hash set since we're consuming all pending events */
+    pending_hash_clear(loop);
+
     pthread_mutex_unlock(&loop->mutex);
 
     return list;

c_src/py_event_loop.h

@@ -49,6 +49,9 @@
 /** @brief Maximum events to keep in freelist (Phase 7 optimization) */
 #define EVENT_FREELIST_SIZE 256
 
+/** @brief Size of pending event hash set for O(1) duplicate detection */
+#define PENDING_HASH_SIZE 128
+
 /** @brief Event types for pending callbacks */
 typedef enum {
     EVENT_TYPE_READ = 1,
@@ -210,6 +213,22 @@ typedef struct erlang_event_loop {
 
     /** @brief Number of events currently in freelist */
     int freelist_count;
+
+    /* ========== O(1) Duplicate Detection Hash Set ========== */
+
+    /**
+     * @brief Hash set for O(1) duplicate pending event detection
+     *
+     * Key: (callback_id, type) combined into a single uint64_t
+     * Uses open addressing with linear probing.
+     */
+    uint64_t pending_hash_keys[PENDING_HASH_SIZE];
+
+    /** @brief Occupancy flags for hash set slots */
+    bool pending_hash_occupied[PENDING_HASH_SIZE];
+
+    /** @brief Count of occupied slots in hash set */
+    int pending_hash_count;
 } erlang_event_loop_t;
 
 /* ============================================================================