1

lib/u_async.c

@@ -1,9 +1,10 @@
 // uasync.c
 
-#include "u_async.h"
-#include "platform_compat.h"
-#include "debug_config.h"
-#include "mem.h"
+#include "u_async.h"
+#include "platform_compat.h"
+#include "debug_config.h"
+#include "mem.h"
+#include "memory_pool.h"
 #include <stdio.h>
 #include <string.h>
 #include <stdlib.h>
@@ -27,13 +28,13 @@
 
 
 
-// Timeout node with safe cancellation
-struct timeout_node {
-    void* arg;
-    timeout_callback_t callback;
-    uint64_t expiration_ms; // absolute expiration time in milliseconds
-    struct UASYNC* ua;      // Pointer back to uasync instance for counter updates
-    int cancelled;          // Cancellation flag
+// Timeout node with safe cancellation
+struct timeout_node {
+    void* arg;
+    timeout_callback_t callback;
+    uint64_t expiration_ms; // absolute expiration time in milliseconds
+    struct UASYNC* ua;      // Pointer back to uasync instance for counter updates
+    struct timeout_node* next;  // For immediate queue (FIFO)
 };
 
 // Socket node with array-based storage
@@ -273,13 +274,13 @@ static struct socket_node* socket_array_get_by_sock(struct socket_array* sa, soc
     return &sa->sockets[index];
 }
 
-// Callback to u_free timeout node and update counters
-static void timeout_node_free_callback(void* user_data, void* data) {
-    struct UASYNC* ua = (struct UASYNC*)user_data;
-    struct timeout_node* node = (struct timeout_node*)data;
-    (void)node; // Not used directly, but keep for consistency
-    ua->timer_free_count++;
-    u_free(data);
+// Callback to u_free timeout node and update counters
+static void timeout_node_free_callback(void* user_data, void* data) {
+    struct UASYNC* ua = (struct UASYNC*)user_data;
+    struct timeout_node* node = (struct timeout_node*)data;
+    (void)node; // Not used directly, but keep for consistency
+    ua->timer_free_count++;
+    memory_pool_free(ua->timeout_pool, data);
 }
 
 // Helper to get current time
@@ -398,35 +399,55 @@ static uint64_t timeval_to_ms(const struct timeval* tv) {
 
 // Simplified timeout handling without reference counting
 
-// Process expired timeouts with safe cancellation
-static void process_timeouts(struct UASYNC* ua) {
-    if (!ua || !ua->timeout_heap) return;
-    
-    struct timeval now_tv;
-    get_current_time(&now_tv);
-    uint64_t now_ms = timeval_to_ms(&now_tv);
-    
-    while (1) {
-        TimeoutEntry entry;
-        if (timeout_heap_peek(ua->timeout_heap, &entry) != 0) break;
-        if (entry.expiration > now_ms) break;
-        
-        // Pop the expired timeout
-        timeout_heap_pop(ua->timeout_heap, &entry);
-        struct timeout_node* node = (struct timeout_node*)entry.data;
-        
-        if (node && node->callback && !node->cancelled) {
-            // Execute callback only if not cancelled
-            node->callback(node->arg);
-        }
-        
-        // Always u_free the node after processing
-        if (node && node->ua) {
-            node->ua->timer_free_count++;
-        }
-        u_free(node);
-        continue;  // Process next expired timeout
-    }
+// Process expired timeouts with safe cancellation
+static void process_timeouts(struct UASYNC* ua) {
+    if (!ua) return;
+    
+    // Сначала обрабатываем immediate_queue (FIFO)
+    while (ua->immediate_queue_head) {
+        struct timeout_node* node = ua->immediate_queue_head;
+        ua->immediate_queue_head = node->next;
+        if (!ua->immediate_queue_head) {
+            ua->immediate_queue_tail = NULL;
+        }
+        
+        if (node && node->callback) {
+            node->callback(node->arg);
+        }
+        
+        if (node && node->ua) {
+            node->ua->timer_free_count++;
+        }
+        memory_pool_free(ua->timeout_pool, node);
+    }
+    
+    if (!ua->timeout_heap) return;
+    
+    struct timeval now_tv;
+    get_current_time(&now_tv);
+    uint64_t now_ms = timeval_to_ms(&now_tv);
+    
+    while (1) {
+        TimeoutEntry entry;
+        if (timeout_heap_peek(ua->timeout_heap, &entry) != 0) break;
+        if (entry.expiration > now_ms) break;
+        
+        // Pop the expired timeout
+        timeout_heap_pop(ua->timeout_heap, &entry);
+        struct timeout_node* node = (struct timeout_node*)entry.data;
+        
+        if (node && node->callback) {
+            // Execute callback only if not cancelled
+            node->callback(node->arg);
+        }
+        
+        // Always u_free the node after processing
+        if (node && node->ua) {
+            node->ua->timer_free_count++;
+        }
+        memory_pool_free(ua->timeout_pool, node);
+        continue;  // Process next expired timeout
+    }
 }
 
 // Compute time to next timeout
@@ -468,17 +489,16 @@ void* uasync_set_timeout(struct UASYNC* ua, int timeout_tb, void* arg, timeout_c
     
 //    DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "uasync_set_timeout: timeout=%d.%d ms, arg=%p, callback=%p", timeout_tb/10, timeout_tb%10, arg, callback);
     
-    struct timeout_node* node = u_malloc(sizeof(struct timeout_node));
-    if (!node) {
-        DEBUG_ERROR(DEBUG_CATEGORY_TIMERS, "uasync_set_timeout: failed to allocate node");
-        return NULL;
+    struct timeout_node* node = memory_pool_alloc(ua->timeout_pool);
+    if (!node) {
+        DEBUG_ERROR(DEBUG_CATEGORY_TIMERS, "uasync_set_timeout: failed to allocate node");
+        return NULL;
     }
     ua->timer_alloc_count++;
     
     node->arg = arg;
     node->callback = callback;
     node->ua = ua;
-    node->cancelled = 0;
     
     // Calculate expiration time in milliseconds
     struct timeval now;
@@ -486,19 +506,61 @@ void* uasync_set_timeout(struct UASYNC* ua, int timeout_tb, void* arg, timeout_c
     timeval_add_tb(&now, timeout_tb);
     node->expiration_ms = timeval_to_ms(&now);
     
-    // Add to heap
-    if (timeout_heap_push(ua->timeout_heap, node->expiration_ms, node) != 0) {
-        DEBUG_ERROR(DEBUG_CATEGORY_TIMERS, "uasync_set_timeout: failed to push to heap");
-        u_free(node);
-        ua->timer_free_count++;  // Balance the alloc counter
-        return NULL;
+    // Add to heap
+    if (timeout_heap_push(ua->timeout_heap, node->expiration_ms, node) != 0) {
+        DEBUG_ERROR(DEBUG_CATEGORY_TIMERS, "uasync_set_timeout: failed to push to heap");
+        memory_pool_free(ua->timeout_pool, node);
+        ua->timer_free_count++;  // Balance the alloc counter
+        return NULL;
     }
     
-    return node;
-}
-
-
-
+    return node;
+}
+
+// Immediate execution in next mainloop (FIFO order)
+void* uasync_call_soon(struct UASYNC* ua, void* user_arg, timeout_callback_t callback) {
+    if (!ua || !callback) return NULL;
+    if (!ua->timeout_pool) return NULL;
+    
+    struct timeout_node* node = memory_pool_alloc(ua->timeout_pool);
+    if (!node) {
+        DEBUG_ERROR(DEBUG_CATEGORY_TIMERS, "uasync_call_soon: failed to allocate node");
+        return NULL;
+    }
+    ua->timer_alloc_count++;
+    
+    node->arg = user_arg;
+    node->callback = callback;
+    node->ua = ua;
+    node->expiration_ms = 0;
+    node->next = NULL;
+    
+    // FIFO: добавляем в конец очереди
+    if (ua->immediate_queue_tail) {
+        ua->immediate_queue_tail->next = node;
+        ua->immediate_queue_tail = node;
+    } else {
+        ua->immediate_queue_head = ua->immediate_queue_tail = node;
+    }
+    
+    return node;
+}
+
+// Cancel immediate callback by setting callback to NULL - O(1)
+err_t uasync_call_soon_cancel(struct UASYNC* ua, void* t_id) {
+    if (!ua || !t_id) return ERR_FAIL;
+    
+    struct timeout_node* node = (struct timeout_node*)t_id;
+    if (node->ua != ua) return ERR_FAIL;
+    
+    // Simply nullify callback - will be skipped in process_timeouts
+    node->callback = NULL;
+    
+    return ERR_OK;
+}
+
+
+
 // Instance version
 err_t uasync_cancel_timeout(struct UASYNC* ua, void* t_id) {
     if (!ua || !t_id || !ua->timeout_heap) {
@@ -512,7 +574,6 @@ err_t uasync_cancel_timeout(struct UASYNC* ua, void* t_id) {
     // Try to cancel from heap first
     if (timeout_heap_cancel(ua->timeout_heap, node->expiration_ms, node) == 0) {
         // Successfully marked as deleted - u_free will happen lazily in heap
-        node->cancelled = 1;
         node->callback = NULL;
 //        DEBUG_DEBUG(DEBUG_CATEGORY_TIMERS, "uasync_cancel_timeout: successfully cancelled timer %p from heap", node);
         return ERR_OK;
@@ -1107,11 +1168,13 @@ struct UASYNC* uasync_create(void) {
     ua->poll_fds_count = 0;
     ua->poll_fds_dirty = 1;
 
-    ua->wakeup_pipe[0] = -1;
-    ua->wakeup_pipe[1] = -1;
-    ua->wakeup_initialized = 0;
-    ua->posted_tasks_head = NULL;
-
+    ua->wakeup_pipe[0] = -1;
+    ua->wakeup_pipe[1] = -1;
+    ua->wakeup_initialized = 0;
+    ua->posted_tasks_head = NULL;
+    ua->immediate_queue_head = NULL;
+    ua->immediate_queue_tail = NULL;
+
     DEBUG_INFO(DEBUG_CATEGORY_ETCP, "Creating SA...");
     ua->sockets = socket_array_create(16);
     DEBUG_INFO(DEBUG_CATEGORY_ETCP, "Creating SA1...");
@@ -1129,22 +1192,40 @@ struct UASYNC* uasync_create(void) {
         return NULL;
     }
 
-    ua->timeout_heap = timeout_heap_create(16);
-    if (!ua->timeout_heap) {
-        socket_array_destroy(ua->sockets);
-        if (ua->wakeup_initialized) {
-#ifdef _WIN32
-            closesocket((SOCKET)(intptr_t)ua->wakeup_pipe[0]);
-            closesocket((SOCKET)(intptr_t)ua->wakeup_pipe[1]);
-#else
-            close(ua->wakeup_pipe[0]);
-            close(ua->wakeup_pipe[1]);
-#endif
-        }
-        u_free(ua);
-        return NULL;
-    }
-
+    ua->timeout_heap = timeout_heap_create(16);
+    if (!ua->timeout_heap) {
+        socket_array_destroy(ua->sockets);
+        if (ua->wakeup_initialized) {
+#ifdef _WIN32
+            closesocket((SOCKET)(intptr_t)ua->wakeup_pipe[0]);
+            closesocket((SOCKET)(intptr_t)ua->wakeup_pipe[1]);
+#else
+            close(ua->wakeup_pipe[0]);
+            close(ua->wakeup_pipe[1]);
+#endif
+        }
+        u_free(ua);
+        return NULL;
+    }
+
+    // Initialize timeout pool
+    ua->timeout_pool = memory_pool_init(sizeof(struct timeout_node));
+    if (!ua->timeout_pool) {
+        timeout_heap_destroy(ua->timeout_heap);
+        socket_array_destroy(ua->sockets);
+        if (ua->wakeup_initialized) {
+#ifdef _WIN32
+            closesocket((SOCKET)(intptr_t)ua->wakeup_pipe[0]);
+            closesocket((SOCKET)(intptr_t)ua->wakeup_pipe[1]);
+#else
+            close(ua->wakeup_pipe[0]);
+            close(ua->wakeup_pipe[1]);
+#endif
+        }
+        u_free(ua);
+        return NULL;
+    }
+
     // Set callback to u_free timeout nodes and update counters
     timeout_heap_set_free_callback(ua->timeout_heap, ua, timeout_node_free_callback);
 
@@ -1256,8 +1337,8 @@ void uasync_print_resources(struct UASYNC* ua, const char* prefix) {
             if (!ua->timeout_heap->heap[i].deleted) {
                 active_timers++;
                 struct timeout_node* node = (struct timeout_node*)ua->timeout_heap->heap[i].data;
-                printf("   Timer: node=%p, expires=%llu ms, cancelled=%d\n",
-                       node, (unsigned long long)ua->timeout_heap->heap[i].expiration, node->cancelled);
+                printf("   Timer: node=%p, expires=%llu ms\n",
+                       node, (unsigned long long)ua->timeout_heap->heap[i].expiration);
             }
         }
         printf("   Active timers in heap: %zu\n", active_timers);
@@ -1306,21 +1387,41 @@ void uasync_destroy(struct UASYNC* ua, int close_fds) {
         // Continue cleanup, will abort after if leaks remain
     }
 
-    // Free all remaining timeouts
-    if (ua->timeout_heap) {
-        size_t u_freed_count = 0;
-        while (1) {
-            TimeoutEntry entry;
-            if (timeout_heap_pop(ua->timeout_heap, &entry) != 0) break;
-            struct timeout_node* node = (struct timeout_node*)entry.data;
-
-            // Free all timer nodes (avoid double-u_free bug)
-            if (node) {
-                ua->timer_free_count++;
-                u_free(node);
-            }
-        }
-        timeout_heap_destroy(ua->timeout_heap);
+    // Free all remaining timeouts
+    
+    // Очистить immediate_queue
+    while (ua->immediate_queue_head) {
+        struct timeout_node* node = ua->immediate_queue_head;
+        ua->immediate_queue_head = node->next;
+        if (node) {
+            node->ua->timer_free_count++;
+            memory_pool_free(ua->timeout_pool, node);
+        }
+    }
+    ua->immediate_queue_tail = NULL;
+    
+    // Очистить heap
+    if (ua->timeout_heap) {
+        size_t u_freed_count = 0;
+        while (1) {
+            TimeoutEntry entry;
+            if (timeout_heap_pop(ua->timeout_heap, &entry) != 0) break;
+            struct timeout_node* node = (struct timeout_node*)entry.data;
+
+            // Free all timer nodes (avoid double-u_free bug)
+            if (node) {
+                ua->timer_free_count++;
+                memory_pool_free(ua->timeout_pool, node);
+            }
+        }
+        timeout_heap_destroy(ua->timeout_heap);
+        ua->timeout_heap = NULL;
+    }
+
+    // Destroy timeout pool
+    if (ua->timeout_pool) {
+        memory_pool_destroy(ua->timeout_pool);
+        ua->timeout_pool = NULL;
     }
 
     // Free all socket nodes using array approach
@@ -1401,20 +1502,29 @@ void uasync_destroy(struct UASYNC* ua, int close_fds) {
     socket_platform_cleanup();
 }
 
-void uasync_init_instance(struct UASYNC* ua) {
-    if (!ua) return;
-    
-    // Initialize socket array if not present
-    if (!ua->sockets) {
-        ua->sockets = socket_array_create(16);
-    }
-    
-    if (!ua->timeout_heap) {
-        ua->timeout_heap = timeout_heap_create(16);
-        if (ua->timeout_heap) {
-            timeout_heap_set_free_callback(ua->timeout_heap, ua, timeout_node_free_callback);
-        }
-    }
+void uasync_init_instance(struct UASYNC* ua) {
+    if (!ua) return;
+    
+    // Initialize socket array if not present
+    if (!ua->sockets) {
+        ua->sockets = socket_array_create(16);
+    }
+    
+    if (!ua->timeout_pool) {
+        ua->timeout_pool = memory_pool_init(sizeof(struct timeout_node));
+    }
+    
+    if (!ua->immediate_queue_head) {
+        ua->immediate_queue_head = NULL;
+        ua->immediate_queue_tail = NULL;
+    }
+    
+    if (!ua->timeout_heap) {
+        ua->timeout_heap = timeout_heap_create(16);
+        if (ua->timeout_heap) {
+            timeout_heap_set_free_callback(ua->timeout_heap, ua, timeout_node_free_callback);
+        }
+    }
 }
 
 // Debug statistics