diff --git a/AGENTS.md b/AGENTS.md index 1ed0434a..18ce3a0f 100755 --- a/AGENTS.md +++ b/AGENTS.md @@ -6,10 +6,15 @@ This document provides essential information for agentic coding assistants worki Если не работает - добавляй отладочную информацию чтобы быстрее найти проблемное место и не рушить работающий код. если отладочная информация будет флудить сделай ее отключаемой или подумай как ограничить ее вывод по возможности сохраняя информативность. Добавляй в код структуры для статистики (например суммируй число ошибок, вызовов и других потенциально нужных для анализа при ошибках метрик). -добавляй в тест подсчет времени сколько каждфй этап длился для лучшего понимания оптимизации +добавляй в тест подсчет времени сколько каждый этап длился для лучшего понимания оптимизации перед запуском тестов не забывай их пересобирать веди changelog.txt: дата время: что поменялось +задача протокола - обеспечивать минимальную задержку при большом трафике. т.е. держать необходимый минимум данных в очередях. +дизайн - однопоточный асинхронный. +все ожидающие операции - через сокеты(дескрипторы) и таймауты -> u_async +все очереди через ll_queue +главный модуль обслуживания подключения - connection ## Code Style Guidelines diff --git a/changelog.txt b/changelog.txt index 2bf49342..81148736 100644 --- a/changelog.txt +++ b/changelog.txt @@ -148,3 +148,13 @@ Fri Jan 16 2026 23:45: Исправление pacing механизма в stres - Добавлена очистка pacing_waiter в callback для избежания висячих указателей - Pacing теперь работает корректно: пакеты отправляются по одному с ожиданием опустошения очереди - Добавлены детальные отладочные логи для отслеживания вызовов и состояния очередей + +Sat Jan 17 2026 00:15: Оптимизация pacing механизма и устранение race condition + + - Устранена критическая race condition: callback больше не пытается отменить уже удаленный waiter + - Упрощена логика управления waiter'ами: callback просто очищает указатель и отправляет следующий пакет + - Добавлена статистика pacing для анализа производительности (immediate callbacks, waiter registrations, cycles) + - Оптимизирована производительность: убраны лишние отладочные сообщения в hot path + - Улучшена обработка NULL от queue_wait_threshold: разделены случаи "очередь пуста" и "ошибка" + - Добавлена условная компиляция для отладочных сообщений в ll_queue.c (LL_QUEUE_DEBUG) + - Pacing теперь работает быстро при нулевых очередях: callback вызывается немедленно при опустошении diff --git a/src/config_parser.c b/src/config_parser.c index 8b277aec..f53f5837 100644 --- a/src/config_parser.c +++ b/src/config_parser.c @@ -10,6 +10,9 @@ #define MAX_LINE_LEN 1024 #define INITIAL_CONNECTION_CAPACITY 4 +#define INITIAL_SERVER_CAPACITY 4 +#define INITIAL_CLIENT_CAPACITY 4 +#define INITIAL_CONNECTION_V2_CAPACITY 4 // Helper function to trim whitespace static char* trim(char *str) { @@ -49,192 +52,119 @@ static int parse_key_value(const char *line, char *key, size_t key_len, return 0; } -// Parse connection mode string -static config_conn_mode_t parse_mode(const char *mode_str) { - if (!mode_str) return CONFIG_MODE_UNKNOWN; - - if (strcasecmp(mode_str, "client") == 0) { - return CONFIG_MODE_CLIENT; - } else if (strcasecmp(mode_str, "server") == 0) { - return CONFIG_MODE_SERVER; - } - - return CONFIG_MODE_UNKNOWN; -} -// Add a connection to the config -static int add_connection(utun_config_t *config, const connection_config_t *conn) { - if (config->connection_count >= config->connection_capacity) { + + + +// Add a server to the config +static int add_server(utun_config_t *config, const server_config_t *server) { + if (config->server_count >= config->server_capacity) { // Resize array - int new_capacity = config->connection_capacity * 2; - connection_config_t *new_connections = realloc(config->connections, - new_capacity * sizeof(connection_config_t)); - if (!new_connections) return -1; + int new_capacity = config->server_capacity * 2; + server_config_t *new_servers = realloc(config->servers, + new_capacity * sizeof(server_config_t)); + if (!new_servers) return -1; - config->connections = new_connections; - config->connection_capacity = new_capacity; + config->servers = new_servers; + config->server_capacity = new_capacity; } - // Copy connection data - memcpy(&config->connections[config->connection_count], conn, sizeof(connection_config_t)); - config->connection_count++; + // Copy server data + memcpy(&config->servers[config->server_count], server, sizeof(server_config_t)); + config->server_count++; return 0; } -// Parse configuration file -utun_config_t* parse_config(const char *filename) { - FILE *fp = fopen(filename, "r"); - if (!fp) { - fprintf(stderr, "Failed to open config file: %s\n", filename); - return NULL; +// Add a client to the config +static int add_client(utun_config_t *config, const client_config_t *client) { + if (config->client_count >= config->client_capacity) { + // Resize array + int new_capacity = config->client_capacity * 2; + client_config_t *new_clients = realloc(config->clients, + new_capacity * sizeof(client_config_t)); + if (!new_clients) return -1; + + config->clients = new_clients; + config->client_capacity = new_capacity; } - // Allocate config structure - utun_config_t *config = calloc(1, sizeof(utun_config_t)); - if (!config) { - fclose(fp); - return NULL; - } + // Copy client data + memcpy(&config->clients[config->client_count], client, sizeof(client_config_t)); + config->client_count++; - // Initialize connection array - config->connection_capacity = INITIAL_CONNECTION_CAPACITY; - config->connections = malloc(config->connection_capacity * sizeof(connection_config_t)); - if (!config->connections) { - free(config); - fclose(fp); - return NULL; + return 0; +} + +// Add a route to connection v2 +static int add_route_to_connection(connection_config_v2_t *conn_v2, const route_pair_t *route) { + if (conn_v2->route_count >= conn_v2->route_capacity) { + // Resize array + int new_capacity = conn_v2->route_capacity == 0 ? INITIAL_CONNECTION_V2_CAPACITY : conn_v2->route_capacity * 2; + route_pair_t *new_routes = realloc(conn_v2->routes, + new_capacity * sizeof(route_pair_t)); + if (!new_routes) return -1; + + conn_v2->routes = new_routes; + conn_v2->route_capacity = new_capacity; } - char line[MAX_LINE_LEN]; - char current_section[128] = ""; - connection_config_t current_conn = {0}; - int in_connection_section = 0; + // Copy route data + memcpy(&conn_v2->routes[conn_v2->route_count], route, sizeof(route_pair_t)); + conn_v2->route_count++; - while (fgets(line, sizeof(line), fp)) { - // Remove newline - line[strcspn(line, "\n")] = '\0'; - - char *trimmed = trim(line); - - // Skip empty lines and comments - if (strlen(trimmed) == 0 || trimmed[0] == ';' || trimmed[0] == '#') { - continue; - } - - // Check for section header - if (trimmed[0] == '[' && trimmed[strlen(trimmed) - 1] == ']') { - // End previous connection section if any - if (in_connection_section) { - if (strlen(current_conn.name) > 0) { - if (add_connection(config, ¤t_conn) != 0) { - fprintf(stderr, "Failed to add connection: %s\n", current_conn.name); - } - } - memset(¤t_conn, 0, sizeof(current_conn)); - in_connection_section = 0; - } - - // Extract section name - strncpy(current_section, trimmed + 1, sizeof(current_section) - 1); - current_section[sizeof(current_section) - 1] = '\0'; - current_section[strcspn(current_section, "]")] = '\0'; - trim(current_section); - - // Check if it's a connection section - if (strncmp(current_section, "connection:", 11) == 0) { - in_connection_section = 1; - // Extract connection name - char *name = trim(current_section + 11); - if (strlen(name) > 0) { - strncpy(current_conn.name, name, sizeof(current_conn.name) - 1); - } - } - - continue; - } - - // Parse key-value pair - char key[256], value[256]; - if (parse_key_value(trimmed, key, sizeof(key), value, sizeof(value)) != 0) { - fprintf(stderr, "Invalid key-value line: %s\n", trimmed); - continue; - } + return 0; +} + +// Add a connection v2 to the config +static int add_connection_v2(utun_config_t *config, const connection_config_v2_t *conn_v2) { + if (config->connection_v2_count >= config->connection_v2_capacity) { + // Resize array + int new_capacity = config->connection_v2_capacity * 2; + connection_config_v2_t *new_connections = realloc(config->connections_v2, + new_capacity * sizeof(connection_config_v2_t)); + if (!new_connections) return -1; - // Process based on current section - if (strcasecmp(current_section, "global") == 0) { - if (strcasecmp(key, "my_private_key") == 0) { - strncpy(config->global.my_private_key_hex, value, sizeof(config->global.my_private_key_hex) - 1); - } else if (strcasecmp(key, "my_public_key") == 0) { - strncpy(config->global.my_public_key_hex, value, sizeof(config->global.my_public_key_hex) - 1); - } else if (strcasecmp(key, "option") == 0) { - strncpy(config->global.option_value, value, sizeof(config->global.option_value) - 1); - } else if (strcasecmp(key, "control_ip") == 0) { - strncpy(config->global.control_ip, value, sizeof(config->global.control_ip) - 1); - } else if (strcasecmp(key, "control_port") == 0) { - config->global.control_port = atoi(value); - } else if (strcasecmp(key, "net_debug") == 0) { - config->global.net_debug = atoi(value); - } - } else if (strcasecmp(current_section, "routing") == 0) { - if (strcasecmp(key, "allowed_subnet") == 0) { - if (config->allowed_subnet_count < MAX_ALLOWED_SUBNETS) { - strncpy(config->allowed_subnets[config->allowed_subnet_count].subnet, value, sizeof(config->allowed_subnets[0].subnet) - 1); - config->allowed_subnet_count++; - } else { - fprintf(stderr, "Too many allowed subnets, maximum is %d\n", MAX_ALLOWED_SUBNETS); - } - } - } else if (in_connection_section) { - if (strcasecmp(key, "mode") == 0) { - current_conn.mode = parse_mode(value); - } else if (strcasecmp(key, "addr") == 0) { - // For server mode - if (current_conn.mode == CONFIG_MODE_SERVER) { - strncpy(current_conn.local_addr, value, sizeof(current_conn.local_addr) - 1); - } - } else if (strcasecmp(key, "from_addr") == 0) { - // For client mode - if (current_conn.mode == CONFIG_MODE_CLIENT) { - strncpy(current_conn.local_addr, value, sizeof(current_conn.local_addr) - 1); - } - } else if (strcasecmp(key, "to_addr") == 0) { - // For client mode - if (current_conn.mode == CONFIG_MODE_CLIENT) { - strncpy(current_conn.remote_addr, value, sizeof(current_conn.remote_addr) - 1); - } - } else if (strcasecmp(key, "peer_public_key") == 0) { - strncpy(current_conn.peer_public_key_hex, value, sizeof(current_conn.peer_public_key_hex) - 1); - } else if (strcasecmp(key, "so_mark") == 0) { - current_conn.so_mark = atoi(value); - } else if (strcasecmp(key, "netif") == 0) { - strncpy(current_conn.netif, value, sizeof(current_conn.netif) - 1); - } else if (strcasecmp(key, "tun") == 0) { - strncpy(current_conn.tun_ifname, value, sizeof(current_conn.tun_ifname) - 1); - } else if (strcasecmp(key, "tun_ip") == 0) { - strncpy(current_conn.tun_ip, value, sizeof(current_conn.tun_ip) - 1); - } - } + config->connections_v2 = new_connections; + config->connection_v2_capacity = new_capacity; } - // Add last connection if any - if (in_connection_section && strlen(current_conn.name) > 0) { - if (add_connection(config, ¤t_conn) != 0) { - fprintf(stderr, "Failed to add connection: %s\n", current_conn.name); - } - } + // Copy connection data + memcpy(&config->connections_v2[config->connection_v2_count], conn_v2, sizeof(connection_config_v2_t)); + config->connection_v2_count++; - fclose(fp); - return config; + return 0; +} + +// Parse configuration file +utun_config_t* parse_config(const char *filename) { + /* Parse only v2 format */ + return parse_config_v2(filename); } // Free configuration structure void free_config(utun_config_t *config) { if (!config) return; - if (config->connections) { - free(config->connections); + + + if (config->servers) { + free(config->servers); } + + if (config->clients) { + free(config->clients); + } + + if (config->connections_v2) { + for (int i = 0; i < config->connection_v2_count; i++) { + if (config->connections_v2[i].routes) { + free(config->connections_v2[i].routes); + } + } + free(config->connections_v2); + } + free(config); } @@ -259,21 +189,39 @@ void print_config(const utun_config_t *config) { } printf("\n"); - printf("Connections (%d):\n", config->connection_count); - for (int i = 0; i < config->connection_count; i++) { - const connection_config_t *conn = &config->connections[i]; - printf(" [%d] %s:\n", i, conn->name); - printf(" mode: %s\n", - conn->mode == CONFIG_MODE_CLIENT ? "client" : - conn->mode == CONFIG_MODE_SERVER ? "server" : "unknown"); - printf(" local_addr: %s\n", conn->local_addr); - printf(" remote_addr: %s\n", conn->remote_addr); + printf("Servers (%d):\n", config->server_count); + for (int i = 0; i < config->server_count; i++) { + const server_config_t *server = &config->servers[i]; + printf(" [%d] %s:\n", i, server->name); + printf(" addr: %s\n", server->addr); + printf(" so_mark: %d\n", server->so_mark); + printf(" netif: %s\n", server->netif); + } + printf("\n"); + + printf("Clients (%d):\n", config->client_count); + for (int i = 0; i < config->client_count; i++) { + const client_config_t *client = &config->clients[i]; + printf(" [%d] %s:\n", i, client->name); + printf(" from: %s\n", client->from); + printf(" to_addr: %s\n", client->to_addr); + } + printf("\n"); + + printf("Connections v2 (%d):\n", config->connection_v2_count); + for (int i = 0; i < config->connection_v2_count; i++) { + const connection_config_v2_t *conn = &config->connections_v2[i]; + printf(" [%d]:\n", i); printf(" peer_public_key: %s\n", conn->peer_public_key_hex); - printf(" so_mark: %d\n", conn->so_mark); - printf(" netif: %s\n", conn->netif); - printf(" tun: %s\n", conn->tun_ifname); - printf(" tun_ip: %s\n", conn->tun_ip); + printf(" keepalive: %d\n", conn->keepalive); + printf(" routes (%d):\n", conn->route_count); + for (int j = 0; j < conn->route_count; j++) { + printf(" %s:%s\n", conn->routes[j].server_name, conn->routes[j].client_name); + } } + printf("\n"); + + } // Update keys in configuration file @@ -298,7 +246,7 @@ int update_config_keys(const char *filename, fclose(fp); return 0; } - + // Read all lines char **lines = NULL; size_t line_count = 0; @@ -410,4 +358,268 @@ int update_config_keys(const char *filename, free(lines); fclose(fp); return 0; +} + +// Parse new format configuration file +utun_config_t* parse_config_v2(const char *filename) { + FILE *fp = fopen(filename, "r"); + if (!fp) { + fprintf(stderr, "Failed to open config file: %s\n", filename); + return NULL; + } + + // Allocate config structure + utun_config_t *config = calloc(1, sizeof(utun_config_t)); + if (!config) { + fclose(fp); + return NULL; + } + + // Initialize arrays + config->server_capacity = INITIAL_SERVER_CAPACITY; + config->servers = malloc(config->server_capacity * sizeof(server_config_t)); + if (!config->servers) { + free(config); + fclose(fp); + return NULL; + } + + config->client_capacity = INITIAL_CLIENT_CAPACITY; + config->clients = malloc(config->client_capacity * sizeof(client_config_t)); + if (!config->clients) { + free(config->servers); + free(config); + fclose(fp); + return NULL; + } + + config->connection_v2_capacity = INITIAL_CONNECTION_V2_CAPACITY; + config->connections_v2 = malloc(config->connection_v2_capacity * sizeof(connection_config_v2_t)); + if (!config->connections_v2) { + free(config->clients); + free(config->servers); + free(config); + fclose(fp); + return NULL; + } + + char line[MAX_LINE_LEN]; + char current_section[128] = ""; + char current_server_name[MAX_CONN_NAME_LEN] = ""; + char current_client_name[MAX_CONN_NAME_LEN] = ""; + server_config_t current_server = {0}; + client_config_t current_client = {0}; + connection_config_v2_t current_conn_v2 = {0}; + int in_server_section = 0; + int in_client_section = 0; + int in_connection_section = 0; + + while (fgets(line, sizeof(line), fp)) { + // Remove newline + line[strcspn(line, "\n")] = '\0'; + + char *trimmed = trim(line); + + // Skip empty lines and comments + if (strlen(trimmed) == 0 || trimmed[0] == ';' || trimmed[0] == '#') { + continue; + } + + // Check for section header + if (trimmed[0] == '[' && trimmed[strlen(trimmed) - 1] == ']') { + // End previous sections if any + if (in_server_section) { + if (strlen(current_server.name) > 0) { + if (add_server(config, ¤t_server) != 0) { + fprintf(stderr, "Failed to add server: %s\n", current_server.name); + } + } + memset(¤t_server, 0, sizeof(current_server)); + in_server_section = 0; + } + if (in_client_section) { + if (strlen(current_client.name) > 0) { + if (add_client(config, ¤t_client) != 0) { + fprintf(stderr, "Failed to add client: %s\n", current_client.name); + } + } + memset(¤t_client, 0, sizeof(current_client)); + in_client_section = 0; + } + if (in_connection_section) { + if (current_conn_v2.route_count > 0) { + if (add_connection_v2(config, ¤t_conn_v2) != 0) { + fprintf(stderr, "Failed to add connection v2\n"); + } + } + memset(¤t_conn_v2, 0, sizeof(current_conn_v2)); + in_connection_section = 0; + } + + // Extract section name + strncpy(current_section, trimmed + 1, sizeof(current_section) - 1); + current_section[sizeof(current_section) - 1] = '\0'; + current_section[strcspn(current_section, "]")] = '\0'; + trim(current_section); + + // Check section type + if (strncmp(current_section, "server:", 7) == 0) { + in_server_section = 1; + char *name = trim(current_section + 7); + if (strlen(name) > 0) { + strncpy(current_server.name, name, sizeof(current_server.name) - 1); + strncpy(current_server_name, name, sizeof(current_server_name) - 1); + } + } else if (strncmp(current_section, "client:", 7) == 0) { + in_client_section = 1; + char *name = trim(current_section + 7); + if (strlen(name) > 0) { + strncpy(current_client.name, name, sizeof(current_client.name) - 1); + strncpy(current_client_name, name, sizeof(current_client_name) - 1); + } + } else if (strcasecmp(current_section, "connection") == 0) { + in_connection_section = 1; + } else if (strcasecmp(current_section, "global") == 0) { + // Already handled by current_section + } else if (strcasecmp(current_section, "routing") == 0) { + // Already handled by current_section + } else { + fprintf(stderr, "Unknown section: %s\n", current_section); + } + + continue; + } + + // Parse key-value pair + char key[256], value[256]; + if (parse_key_value(trimmed, key, sizeof(key), value, sizeof(value)) != 0) { + fprintf(stderr, "Invalid key-value line: %s\n", trimmed); + continue; + } + + // Process based on current section + if (strcasecmp(current_section, "global") == 0) { + if (strcasecmp(key, "my_private_key") == 0) { + strncpy(config->global.my_private_key_hex, value, sizeof(config->global.my_private_key_hex) - 1); + } else if (strcasecmp(key, "my_public_key") == 0) { + strncpy(config->global.my_public_key_hex, value, sizeof(config->global.my_public_key_hex) - 1); + } else if (strcasecmp(key, "option") == 0) { + strncpy(config->global.option_value, value, sizeof(config->global.option_value) - 1); + } else if (strcasecmp(key, "control_ip") == 0) { + strncpy(config->global.control_ip, value, sizeof(config->global.control_ip) - 1); + } else if (strcasecmp(key, "control_port") == 0) { + config->global.control_port = atoi(value); + } else if (strcasecmp(key, "net_debug") == 0) { + config->global.net_debug = atoi(value); + } else if (strcasecmp(key, "tun_ip") == 0) { + strncpy(config->global.tun_ip, value, sizeof(config->global.tun_ip) - 1); + } + } else if (strcasecmp(current_section, "routing") == 0) { + if (strcasecmp(key, "allowed_subnet") == 0) { + if (config->allowed_subnet_count < MAX_ALLOWED_SUBNETS) { + strncpy(config->allowed_subnets[config->allowed_subnet_count].subnet, value, sizeof(config->allowed_subnets[0].subnet) - 1); + config->allowed_subnet_count++; + } else { + fprintf(stderr, "Too many allowed subnets, maximum is %d\n", MAX_ALLOWED_SUBNETS); + } + } + } else if (in_server_section) { + if (strcasecmp(key, "addr") == 0) { + strncpy(current_server.addr, value, sizeof(current_server.addr) - 1); + } else if (strcasecmp(key, "so_mark") == 0) { + current_server.so_mark = atoi(value); + } else if (strcasecmp(key, "netif") == 0) { + strncpy(current_server.netif, value, sizeof(current_server.netif) - 1); + } + } else if (in_client_section) { + if (strcasecmp(key, "from") == 0) { + strncpy(current_client.from, value, sizeof(current_client.from) - 1); + } else if (strcasecmp(key, "to_addr") == 0) { + strncpy(current_client.to_addr, value, sizeof(current_client.to_addr) - 1); + } + } else if (in_connection_section) { + if (strcasecmp(key, "link") == 0) { + // Parse comma-separated route pairs: server1:client1,server2:client2 + char *token = strtok(value, ","); + while (token != NULL) { + char *pair = trim(token); + char *colon = strchr(pair, ':'); + if (colon) { + route_pair_t route = {0}; + size_t server_len = colon - pair; + if (server_len >= sizeof(route.server_name)) server_len = sizeof(route.server_name) - 1; + strncpy(route.server_name, pair, server_len); + route.server_name[server_len] = '\0'; + trim(route.server_name); + + char *client_name = colon + 1; + strncpy(route.client_name, client_name, sizeof(route.client_name) - 1); + trim(route.client_name); + + if (add_route_to_connection(¤t_conn_v2, &route) != 0) { + fprintf(stderr, "Failed to add route: %s:%s\n", route.server_name, route.client_name); + } + } else { + fprintf(stderr, "Invalid route pair format: %s, expected 'server:client'\n", pair); + } + token = strtok(NULL, ","); + } + } else if (strcasecmp(key, "keepalive") == 0) { + current_conn_v2.keepalive = atoi(value); + } else if (strcasecmp(key, "peer_public_key") == 0) { + strncpy(current_conn_v2.peer_public_key_hex, value, sizeof(current_conn_v2.peer_public_key_hex) - 1); + } + } + } + + // Add last sections if any + if (in_server_section && strlen(current_server.name) > 0) { + if (add_server(config, ¤t_server) != 0) { + fprintf(stderr, "Failed to add server: %s\n", current_server.name); + } + } + if (in_client_section && strlen(current_client.name) > 0) { + if (add_client(config, ¤t_client) != 0) { + fprintf(stderr, "Failed to add client: %s\n", current_client.name); + } + } + if (in_connection_section && current_conn_v2.route_count > 0) { + if (add_connection_v2(config, ¤t_conn_v2) != 0) { + fprintf(stderr, "Failed to add connection v2\n"); + } + } + + fclose(fp); + + // Validate references + for (int i = 0; i < config->connection_v2_count; i++) { + connection_config_v2_t *conn = &config->connections_v2[i]; + for (int j = 0; j < conn->route_count; j++) { + route_pair_t *route = &conn->routes[j]; + // Check server exists + int server_found = 0; + for (int k = 0; k < config->server_count; k++) { + if (strcmp(config->servers[k].name, route->server_name) == 0) { + server_found = 1; + break; + } + } + if (!server_found) { + fprintf(stderr, "Warning: server '%s' referenced in route not found\n", route->server_name); + } + // Check client exists + int client_found = 0; + for (int k = 0; k < config->client_count; k++) { + if (strcmp(config->clients[k].name, route->client_name) == 0) { + client_found = 1; + break; + } + } + if (!client_found) { + fprintf(stderr, "Warning: client '%s' referenced in route not found\n", route->client_name); + } + } + } + + return config; } \ No newline at end of file diff --git a/src/config_parser.h b/src/config_parser.h index 6f776b30..b14e636e 100644 --- a/src/config_parser.h +++ b/src/config_parser.h @@ -4,7 +4,6 @@ #include #include -#include "connection.h" #ifdef __cplusplus extern "C" { @@ -18,30 +17,44 @@ extern "C" { #define MAX_OPTION_VALUE_LEN 256 #define MAX_ALLOWED_SUBNETS 32 -// Connection mode (config parser internal) -typedef enum { - CONFIG_MODE_UNKNOWN = 0, - CONFIG_MODE_CLIENT, - CONFIG_MODE_SERVER -} config_conn_mode_t; + // Subnet entry for allowed routing typedef struct { char subnet[MAX_ADDR_LEN]; // Format: "192.168.0.0/24" } subnet_entry_t; -// Connection configuration + + +// Server configuration (new format) +typedef struct { + char name[MAX_CONN_NAME_LEN]; + char addr[MAX_ADDR_LEN]; + int so_mark; + char netif[MAX_NETIF_LEN]; +} server_config_t; + +// Client configuration (new format) typedef struct { - char name[MAX_CONN_NAME_LEN]; // Connection name from [connection: name] - config_conn_mode_t mode; // server or client - char local_addr[MAX_ADDR_LEN]; // For client: from_addr, for server: addr - char remote_addr[MAX_ADDR_LEN]; // For client: to_addr, for server: N/A - char peer_public_key_hex[MAX_KEY_LEN]; // Peer's public key in HEX - int so_mark; // Socket mark (0 = not set) - char netif[MAX_NETIF_LEN]; // Network interface to bind to - char tun_ifname[MAX_NETIF_LEN]; // TUN interface name (e.g., tun12) - char tun_ip[MAX_ADDR_LEN]; // TUN interface IP address (e.g., 10.0.0.1/24) -} connection_config_t; + char name[MAX_CONN_NAME_LEN]; + char from[MAX_CONN_NAME_LEN]; // server name + char to_addr[MAX_ADDR_LEN]; +} client_config_t; + +// Route pair (server:client) +typedef struct { + char server_name[MAX_CONN_NAME_LEN]; + char client_name[MAX_CONN_NAME_LEN]; +} route_pair_t; + +// Connection configuration v2 (new format) +typedef struct { + route_pair_t *routes; + int route_count; + int route_capacity; + char peer_public_key_hex[MAX_KEY_LEN]; + int keepalive; +} connection_config_v2_t; // Global configuration typedef struct { @@ -51,39 +64,66 @@ typedef struct { char control_ip[MAX_ADDR_LEN]; // Control socket IP (empty for disabled) uint16_t control_port; // Control socket port int net_debug; // Network debug mode: 1 = enable network emulator + char tun_ip[MAX_ADDR_LEN]; // TUN interface IP address with prefix (e.g., 10.0.0.1/24) } global_config_t; // Complete configuration typedef struct { global_config_t global; - connection_config_t *connections; - int connection_count; - int connection_capacity; + + + + // New format configurations + server_config_t *servers; + int server_count; + int server_capacity; + + client_config_t *clients; + int client_count; + int client_capacity; + + connection_config_v2_t *connections_v2; + int connection_v2_count; + int connection_v2_capacity; + subnet_entry_t allowed_subnets[MAX_ALLOWED_SUBNETS]; int allowed_subnet_count; } utun_config_t; /** - * @brief Parse configuration file + * @brief Parse configuration file (new format only) + * @param filename Path to configuration file + * @return Pointer to parsed configuration, NULL on error + */ +utun_config_t* parse_config(const char *filename); + +/** + * @brief Parse new format configuration file * @param filename Path to configuration file * @return Pointer to parsed configuration, NULL on error * - * Parses INI-style configuration file with format: + * Parses new format configuration file: * [global] * option=value * my_private_key=HEX * my_public_key=HEX + * tun_ip=1.2.3.4 + * + * [server: wired1_fast] + * addr=192.168.0.10:1234 + * so_mark=100 + * netif=lo0 * - * [connection: name] - * mode=server|client - * addr=ip:port (for server) - * from_addr=ip:port (for client) - * to_addr=ip:port (for client) + * [client: wireless] + * from=wired1_fast + * to_addr=192.168.0.20:1234 + * + * [connection] + * link=wired1_fast:wireless + * keepalive=1 * peer_public_key=HEX - * so_mark=number - * netif=interface_name */ -utun_config_t* parse_config(const char *filename); +utun_config_t* parse_config_v2(const char *filename); /** * @brief Free configuration structure diff --git a/src/control_socket.c b/src/control_socket.c index f3af348d..3d2cb7f2 100644 --- a/src/control_socket.c +++ b/src/control_socket.c @@ -343,15 +343,9 @@ static int collect_statistics(control_socket_t *cs, control_stats_packet_t *stat // Additional per-connection stats could be stored in conn_stats_entry_t } - // Fill connection entry (if we have array) - if (i < MAX_CONN_STATS) { - if (state->config && state->config->connections) { - strncpy(stats->connections[i].name, state->config->connections[i].name, - sizeof(stats->connections[i].name) - 1); - stats->connections[i].name[sizeof(stats->connections[i].name) - 1] = '\0'; - } else { + // Fill connection entry (if we have array) + if (i < MAX_CONN_STATS) { snprintf(stats->connections[i].name, sizeof(stats->connections[i].name), "conn%d", i); - } stats->connections[i].bytes_sent = conn_stats.bytes_sent; stats->connections[i].bytes_received = conn_stats.bytes_received; stats->connections[i].packets_sent = conn_stats.packets_sent; diff --git a/src/ll_queue.c b/src/ll_queue.c index edf9c0d4..c380d629 100644 --- a/src/ll_queue.c +++ b/src/ll_queue.c @@ -12,8 +12,10 @@ static void queue_resume_timeout_cb(void* arg); static void check_waiters(ll_queue_t* q) { if (!q || !q->waiters) return; +#ifdef LL_QUEUE_DEBUG printf("[LL_QUEUE DEBUG] check_waiters: checking %d waiters, count=%d, bytes=%zu\n", (q->waiters ? 1 : 0), q->count, q->total_bytes); +#endif queue_waiter_t** pprev = &q->waiters; queue_waiter_t* waiter = q->waiters; @@ -23,8 +25,10 @@ static void check_waiters(ll_queue_t* q) { // Проверить условие: не больше max_packets и не больше max_bytes if (q->count <= waiter->max_packets && q->total_bytes <= waiter->max_bytes) { +#ifdef LL_QUEUE_DEBUG printf("[LL_QUEUE DEBUG] check_waiters: condition met, calling callback, count=%d<=%d, bytes=%zu<=%zu\n", q->count, waiter->max_packets, q->total_bytes, waiter->max_bytes); +#endif waiter->callback(q, waiter->callback_arg); // Удалить waiter из списка *pprev = next; @@ -32,8 +36,10 @@ static void check_waiters(ll_queue_t* q) { // pprev уже указывает на правильный следующий элемент } else { // Условие не выполнено - оставить в списке +#ifdef LL_QUEUE_DEBUG printf("[LL_QUEUE DEBUG] check_waiters: condition NOT met, count=%d>%d or bytes=%zu>%zu\n", q->count, waiter->max_packets, q->total_bytes, waiter->max_bytes); +#endif pprev = &waiter->next; } waiter = next; @@ -260,22 +266,28 @@ queue_waiter_t* queue_wait_threshold(ll_queue_t* q, int max_packets, size_t max_ // Проверить условие немедленно if (q->count <= max_packets && q->total_bytes <= max_bytes) { // Условие уже выполнено - вызвать коллбэк и освободить waiter +#ifdef LL_QUEUE_DEBUG printf("[LL_QUEUE DEBUG] queue_wait_threshold: condition already met, count=%d<=%d, bytes=%zu<=%zu, calling callback\n", q->count, max_packets, q->total_bytes, max_bytes); +#endif callback(q, arg); free(waiter); return NULL; } +#ifdef LL_QUEUE_DEBUG printf("[LL_QUEUE DEBUG] queue_wait_threshold: registering waiter, count=%d, bytes=%zu, max_packets=%d, max_bytes=%zu\n", q->count, q->total_bytes, max_packets, max_bytes); +#endif // Добавить в список ожидающих waiter->next = q->waiters; q->waiters = waiter; +#ifdef LL_QUEUE_DEBUG printf("[LL_QUEUE DEBUG] queue_wait_threshold: waiter registered successfully, waiters list=%p, returning waiter=%p\n", (void*)q->waiters, (void*)waiter); +#endif return waiter; } diff --git a/src/ll_queue.h b/src/ll_queue.h index 00cabaaf..d45ee70e 100644 --- a/src/ll_queue.h +++ b/src/ll_queue.h @@ -121,7 +121,7 @@ static inline size_t ll_entry_size(ll_entry_t* entry) { // ==================== Асинхронное ожидание ==================== -// Зарегистрировать коллбэк, который будет вызван когда очередь будет иметь +// Зарегистрировать одноразовый коллбэк, который будет вызван когда очередь будет иметь // не более max_packets пакетов и не более max_bytes байт. // Если условие уже выполнено, коллбэк вызывается немедленно. // Можно зарегистрировать несколько ожиданий на одной очереди. diff --git a/src/utun.c b/src/utun.c index d0cf72a2..094cd190 100644 --- a/src/utun.c +++ b/src/utun.c @@ -18,10 +18,13 @@ #include #include #include -#include + #include #include +// Global wakeup pipe write fd for signal handler +static int g_wakeup_pipe_write_fd = -1; + #define MAX_PACKET_SIZE 2048 #define DEFAULT_CONFIG "utun.conf" #define DEFAULT_PIDFILE "/var/run/utun.pid" @@ -39,44 +42,9 @@ typedef struct { // Global state -// Hex string to binary conversion -static int hex_to_bin(const char *hex, uint8_t *bin, size_t bin_len) { - if (!hex || !bin) return -1; - - size_t hex_len = strlen(hex); - if (hex_len % 2 != 0 || hex_len / 2 > bin_len) return -1; - - for (size_t i = 0; i < hex_len; i += 2) { - char byte_str[3] = {hex[i], hex[i + 1], '\0'}; - char *endptr; - long byte = strtol(byte_str, &endptr, 16); - if (*endptr != '\0') return -1; - bin[i / 2] = (uint8_t)byte; - } - - return hex_len / 2; -} -// Parse IP:port string -static int parse_addr(const char *addr_str, char *ip, size_t ip_len, uint16_t *port) { - if (!addr_str || !ip || !port) return -1; - - char *colon = strchr(addr_str, ':'); - if (!colon) return -1; - - size_t ip_size = colon - addr_str; - if (ip_size >= ip_len) return -1; - - strncpy(ip, addr_str, ip_size); - ip[ip_size] = '\0'; - - char *endptr; - long port_num = strtol(colon + 1, &endptr, 10); - if (*endptr != '\0' || port_num < 1 || port_num > 65535) return -1; - - *port = (uint16_t)port_num; - return 0; -} + + // Parse subnet string static int parse_subnet(const char *subnet_str, uint32_t *network, uint8_t *prefix_length) { @@ -104,11 +72,13 @@ static uint32_t get_dest_ip(const uint8_t *packet, size_t len) { return dest_ip; } -// Initialize connection from configuration -static conn_handle_t* init_connection(uasync_t *ua, - const connection_config_t *conn_cfg, - const global_config_t *global_cfg) { - if (!conn_cfg || !global_cfg) return NULL; +// Initialize connection from v2 configuration +static conn_handle_t* init_connection_v2(uasync_t *ua, + const utun_config_t *config, + int conn_idx) { + if (!config || conn_idx < 0 || conn_idx >= config->connection_v2_count) { + return NULL; + } // Create connection conn_handle_t *conn = conn_create(ua); @@ -117,85 +87,9 @@ static conn_handle_t* init_connection(uasync_t *ua, return NULL; } - // Convert keys from hex to binary - uint8_t my_priv_key[32] = {0}; - uint8_t my_pub_key[64] = {0}; - uint8_t peer_pub_key[64] = {0}; - - if (strlen(global_cfg->my_private_key_hex) > 0) { - if (hex_to_bin(global_cfg->my_private_key_hex, my_priv_key, sizeof(my_priv_key)) < 0) { - fprintf(stderr, "Invalid private key format\n"); - conn_destroy(conn); - return NULL; - } - } - - if (strlen(global_cfg->my_public_key_hex) > 0) { - if (hex_to_bin(global_cfg->my_public_key_hex, my_pub_key, sizeof(my_pub_key)) < 0) { - fprintf(stderr, "Invalid public key format\n"); - conn_destroy(conn); - return NULL; - } - } - - if (strlen(conn_cfg->peer_public_key_hex) > 0) { - if (hex_to_bin(conn_cfg->peer_public_key_hex, peer_pub_key, sizeof(peer_pub_key)) < 0) { - fprintf(stderr, "Invalid peer public key format\n"); - conn_destroy(conn); - return NULL; - } - } - - // Set keys - if (conn_set_keys(conn, my_pub_key, my_priv_key, peer_pub_key) < 0) { - fprintf(stderr, "Failed to set keys\n"); - conn_destroy(conn); - return NULL; - } - - // Parse addresses - char local_ip[64] = ""; - uint16_t local_port = 0; - char remote_ip[64] = ""; - uint16_t remote_port = 0; - - if (strlen(conn_cfg->local_addr) > 0) { - if (parse_addr(conn_cfg->local_addr, local_ip, sizeof(local_ip), &local_port) < 0) { - fprintf(stderr, "Invalid local address format: %s\n", conn_cfg->local_addr); - conn_destroy(conn); - return NULL; - } - } - - if (conn_cfg->mode == CONFIG_MODE_CLIENT && strlen(conn_cfg->remote_addr) > 0) { - if (parse_addr(conn_cfg->remote_addr, remote_ip, sizeof(remote_ip), &remote_port) < 0) { - fprintf(stderr, "Invalid remote address format: %s\n", conn_cfg->remote_addr); - conn_destroy(conn); - return NULL; - } - - // Apply net_debug mode: redirect traffic to emulator port (+10000) - if (global_cfg->net_debug && remote_port > 0) { - uint16_t original_port = remote_port; - remote_port += 10000; - if (remote_port < original_port) { // Overflow check - remote_port = 65535; - } - printf("[NET_DEBUG] Redirecting connection %s:%u -> %s:%u (emulator port %u)\n", - conn_cfg->name, original_port, remote_ip, remote_port - 10000, remote_port); - } - } - - // Connect - conn_mode_t mode = (conn_cfg->mode == CONFIG_MODE_CLIENT) ? CONN_MODE_CLIENT : CONN_MODE_SERVER; - - if (conn_connect(conn, - strlen(local_ip) > 0 ? local_ip : NULL, - local_port, - strlen(remote_ip) > 0 ? remote_ip : NULL, - remote_port, - mode) < 0) { - fprintf(stderr, "Failed to connect\n"); + // Initialize connection using v2 API + if (conn_init_v2(conn, config, conn_idx) < 0) { + fprintf(stderr, "Failed to initialize connection v2\n"); conn_destroy(conn); return NULL; } @@ -203,6 +97,8 @@ static conn_handle_t* init_connection(uasync_t *ua, return conn; } + + // Callback for received data from connection static void connection_recv_callback(conn_handle_t* conn, const uint8_t* data, @@ -225,20 +121,23 @@ static void connection_recv_callback(conn_handle_t* conn, } } -// Initialize all connections +// Initialize all connections from v2 configuration static int init_connections(utun_state_t *state) { if (!state || !state->config) return -1; - state->connection_count = state->config->connection_count; + state->connection_count = state->config->connection_v2_count; + if (state->connection_count == 0) { + fprintf(stderr, "No v2 connections found in configuration\n"); + return -1; + } + state->connections = calloc(state->connection_count, sizeof(conn_handle_t*)); if (!state->connections) return -1; for (int i = 0; i < state->connection_count; i++) { - state->connections[i] = init_connection(state->ua, - &state->config->connections[i], - &state->config->global); + state->connections[i] = init_connection_v2(state->ua, state->config, i); if (!state->connections[i]) { - fprintf(stderr, "Failed to initialize connection %d\n", i); + fprintf(stderr, "Failed to initialize v2 connection %d\n", i); // Cleanup already created connections for (int j = 0; j < i; j++) { conn_destroy(state->connections[j]); @@ -458,6 +357,7 @@ static void cleanup(utun_state_t *state, const char *pidfile) { if (state->ua) { uasync_destroy(state->ua); state->ua = NULL; + g_wakeup_pipe_write_fd = -1; } // Close TUN device @@ -484,6 +384,12 @@ static volatile sig_atomic_t got_signal = 0; static void signal_handler(int sig) { (void)sig; // unused got_signal = 1; + + // Write to wakeup pipe to interrupt poll + if (g_wakeup_pipe_write_fd >= 0) { + char byte = 0; + (void)write(g_wakeup_pipe_write_fd, &byte, 1); // async-signal-safe, ignore result + } } // Setup signal handlers @@ -498,89 +404,78 @@ static void setup_signals(void) { sigaction(SIGHUP, &sa, NULL); } +// Callback for TUN device read events +static void tun_read_callback(int fd, void* user_arg) { + utun_state_t* state = (utun_state_t*)user_arg; + uint8_t buffer[MAX_PACKET_SIZE]; + + // Read from TUN device + ssize_t nread = tun_read(fd, buffer, sizeof(buffer)); + if (nread < 0) { + if (errno == EINTR) return; + perror("read from TUN"); + state->tun.read_errors++; + state->running = 0; // Stop event loop on error + return; + } + + if (nread > 0) { + state->tun.bytes_read += nread; + state->tun.packets_read++; + // Route packet based on destination IP + uint32_t dest_ip = get_dest_ip(buffer, nread); + route_entry_t route; + if (dest_ip != 0 && routing_table_lookup(state->routing_table, dest_ip, &route)) { + // Found route, send to next hop connection + if (route.next_hop) { + if (conn_send(route.next_hop, buffer, nread) < 0) { + fprintf(stderr, "Failed to send packet via route\n"); + } + } else { + // Local route, no forwarding needed + } + } else { + // No route found, drop packet + char ip_str[16]; + ip_to_string(dest_ip, ip_str); + fprintf(stderr, "No route for destination IP %s\n", ip_str); + } + } +} + +// Callback for control socket read events +static void control_socket_callback(int fd, void* user_arg) { + (void)fd; // unused + utun_state_t* state = (utun_state_t*)user_arg; + if (control_socket_process(state->control_socket, state) < 0) { + fprintf(stderr, "Error processing control socket request\n"); + } +} + // Main event loop static int event_loop(utun_state_t *state) { - // Determine number of file descriptors to poll - int num_fds = 1; // TUN device always - int ctrl_fd_idx = -1; - - if (state->control_socket) { - num_fds = 2; - ctrl_fd_idx = 1; + // Register TUN file descriptor with u_async + if (state->tun.fd >= 0) { + uasync_add_socket(state->ua, state->tun.fd, tun_read_callback, NULL, NULL, state); + } else { + fprintf(stderr, "TUN file descriptor invalid\n"); + return -1; } - struct pollfd fds[2]; - uint8_t buffer[MAX_PACKET_SIZE]; - - // Setup poll for TUN device - fds[0].fd = state->tun.fd; - fds[0].events = POLLIN; - - // Setup poll for control socket if exists + // Register control socket file descriptor if exists if (state->control_socket) { - fds[ctrl_fd_idx].fd = control_socket_get_fd(state->control_socket); - fds[ctrl_fd_idx].events = POLLIN; + int ctrl_fd = control_socket_get_fd(state->control_socket); + if (ctrl_fd >= 0) { + uasync_add_socket(state->ua, ctrl_fd, control_socket_callback, NULL, NULL, state); + } else { + fprintf(stderr, "Control socket file descriptor invalid\n"); + } } while (state->running && !got_signal) { - // Process async events (timers, sockets) - if (state->ua) { - uasync_poll(state->ua, 0); - } - // Poll all file descriptors with timeout - int ret = poll(fds, num_fds, 100); // 100ms timeout - if (ret < 0) { - if (errno == EINTR) continue; - perror("poll"); - break; - } - - if (ret == 0) { - // Timeout - continue - continue; - } - - // Check control socket first - if (state->control_socket && (fds[ctrl_fd_idx].revents & POLLIN)) { - if (control_socket_process(state->control_socket, state) < 0) { - fprintf(stderr, "Error processing control socket request\n"); - } - } - - // Check TUN device - if (fds[0].revents & POLLIN) { - // Read from TUN device - ssize_t nread = tun_read(state->tun.fd, buffer, sizeof(buffer)); - if (nread < 0) { - if (errno == EINTR) continue; - perror("read from TUN"); - state->tun.read_errors++; - break; - } - - if (nread > 0) { - state->tun.bytes_read += nread; - state->tun.packets_read++; - // Route packet based on destination IP - uint32_t dest_ip = get_dest_ip(buffer, nread); - route_entry_t route; - if (dest_ip != 0 && routing_table_lookup(state->routing_table, dest_ip, &route)) { - // Found route, send to next hop connection - if (route.next_hop) { - if (conn_send(route.next_hop, buffer, nread) < 0) { - fprintf(stderr, "Failed to send packet via route\n"); - } - } else { - // Local route, no forwarding needed - } - } else { - // No route found, drop packet - char ip_str[16]; - ip_to_string(dest_ip, ip_str); - fprintf(stderr, "No route for destination IP %s\n", ip_str); - } - } - } + // Process all async events (timers, sockets, TUN, control socket) + uasync_poll(state->ua, 1000); // 100ms timeout (1000 timebase units = 100ms) + // uasync_poll returns void, errors are handled internally } return 0; @@ -634,6 +529,12 @@ int main(int argc, char *argv[]) { } uasync_init_instance(state.ua); + // Get wakeup pipe write fd for signal handler + g_wakeup_pipe_write_fd = uasync_get_wakeup_fd(state.ua); + if (g_wakeup_pipe_write_fd < 0) { + fprintf(stderr, "Warning: wakeup pipe not available, shutdown may be delayed\n"); + } + // Setup TUN configuration from command line or config if (args.tun_ifname) { strncpy(state.tun.ifname, args.tun_ifname, sizeof(state.tun.ifname) - 1); @@ -642,16 +543,9 @@ int main(int argc, char *argv[]) { strncpy(state.tun.ip_addr, args.tun_ip, sizeof(state.tun.ip_addr) - 1); } - // If not set by command line, try to get from first connection config - if (state.tun.ifname[0] == '\0' && config->connection_count > 0) { - if (config->connections[0].tun_ifname[0] != '\0') { - strncpy(state.tun.ifname, config->connections[0].tun_ifname, sizeof(state.tun.ifname) - 1); - } - } - if (state.tun.ip_addr[0] == '\0' && config->connection_count > 0) { - if (config->connections[0].tun_ip[0] != '\0') { - strncpy(state.tun.ip_addr, config->connections[0].tun_ip, sizeof(state.tun.ip_addr) - 1); - } + // If not set by command line, try to get from global config + if (state.tun.ip_addr[0] == '\0' && config->global.tun_ip[0] != '\0') { + strncpy(state.tun.ip_addr, config->global.tun_ip, sizeof(state.tun.ip_addr) - 1); } state.tun.mtu = 1500; // Default MTU diff --git a/u_async/u_async.c b/u_async/u_async.c index 38ffa91a..f1f0e8e3 100644 --- a/u_async/u_async.c +++ b/u_async/u_async.c @@ -7,10 +7,11 @@ #include #include #include +#include +#include +#include + -#ifndef FD_SETSIZE -#define FD_SETSIZE 1024 // Assume standard size; adjust if needed for your platform -#endif // Timeout node struct timeout_node { @@ -34,16 +35,14 @@ struct socket_node { struct uasync_s { TimeoutHeap* timeout_heap; // Heap for timeout management struct socket_node* socket_head; - int max_fd; - fd_set master_readfds; - fd_set master_writefds; - fd_set master_exceptfds; - struct socket_node* fd_to_node[FD_SETSIZE]; // Debug counters for memory allocation tracking size_t timer_alloc_count; size_t timer_free_count; size_t socket_alloc_count; size_t socket_free_count; + // Wakeup pipe for interrupting poll + int wakeup_pipe[2]; // [0] read, [1] write + int wakeup_initialized; }; // No global instance - each module must use its own uasync_t instance @@ -64,6 +63,17 @@ static void get_current_time(struct timeval* tv) { +// Drain wakeup pipe - read all available bytes +static void drain_wakeup_pipe(uasync_t* ua) { + if (!ua || !ua->wakeup_initialized) return; + + char buf[64]; + while (1) { + ssize_t n = read(ua->wakeup_pipe[0], buf, sizeof(buf)); + if (n <= 0) break; + } +} + // Helper to add timeval: tv += dt (timebase units) static void timeval_add_tb(struct timeval* tv, int dt) { tv->tv_usec += (dt % 10000) * 100; @@ -204,15 +214,9 @@ void* uasync_add_socket(uasync_t* ua, int fd, socket_callback_t read_cbk, socket node->next = ua->socket_head; ua->socket_head = node; - // Update masters (point 1) - if (read_cbk) FD_SET(fd, &ua->master_readfds); - if (write_cbk) FD_SET(fd, &ua->master_writefds); - if (except_cbk) FD_SET(fd, &ua->master_exceptfds); - - // Update map (point 2) - ua->fd_to_node[fd] = node; - - if (fd > ua->max_fd) ua->max_fd = fd; + // No FD_SET needed for poll + // No fd_to_node map needed + // No max_fd needed for poll return node; } @@ -235,24 +239,8 @@ err_t uasync_remove_socket(uasync_t* ua, void* s_id) { ua->socket_head = cur->next; } - // Update masters (point 1) - if (node->read_cbk) FD_CLR(node->fd, &ua->master_readfds); - if (node->write_cbk) FD_CLR(node->fd, &ua->master_writefds); - if (node->except_cbk) FD_CLR(node->fd, &ua->master_exceptfds); - - // Update map (point 2) - ua->fd_to_node[node->fd] = NULL; - ua->socket_free_count++; free(cur); - - // Update max_fd (simple rescan; optimize if needed by checking if removed == max_fd) - ua->max_fd = -1; - cur = ua->socket_head; - while (cur) { - if (cur->fd > ua->max_fd) ua->max_fd = cur->fd; - cur = cur->next; - } return ERR_OK; } prev = cur; @@ -276,59 +264,164 @@ void uasync_poll(uasync_t* ua, int timeout_tb) { /* Process expired timeouts */ process_timeouts(ua); - /* Prepare select with copies of masters */ - fd_set readfds = ua->master_readfds; - fd_set writefds = ua->master_writefds; - fd_set exceptfds = ua->master_exceptfds; - + /* Compute timeout for poll in milliseconds */ + int timeout_ms = -1; // infinite by default + + // Get next timeout from heap struct timeval tv; get_next_timeout(ua, &tv); + if (tv.tv_sec > 0 || tv.tv_usec > 0 || (ua->timeout_heap && ua->timeout_heap->size > 0)) { + // Convert timeval to milliseconds, cap at INT_MAX + uint64_t ms = (uint64_t)tv.tv_sec * 1000ULL + (uint64_t)tv.tv_usec / 1000ULL; + if (ms > INT_MAX) ms = INT_MAX; + timeout_ms = (int)ms; + } + /* If timeout_tb >= 0, compute timeout as min(timeout_tb, existing timer) */ if (timeout_tb >= 0) { - struct timeval user_tv; - user_tv.tv_sec = timeout_tb / 10000; - user_tv.tv_usec = (timeout_tb % 10000) * 100; + // Convert timebase (0.1 ms) to milliseconds + int user_timeout_ms = timeout_tb / 10; + if (timeout_tb % 10 != 0) user_timeout_ms++; // round up - /* If no internal timer or user timeout is smaller */ - if (tv.tv_sec == 0 && tv.tv_usec == 0 && (!ua->timeout_heap || ua->timeout_heap->size == 0)) { - tv = user_tv; - } else if (user_tv.tv_sec < tv.tv_sec || - (user_tv.tv_sec == tv.tv_sec && user_tv.tv_usec < tv.tv_usec)) { - tv = user_tv; + if (timeout_ms < 0 || user_timeout_ms < timeout_ms) { + timeout_ms = user_timeout_ms; } } - struct timeval* ptv = (tv.tv_sec == 0 && tv.tv_usec == 0 && (!ua->timeout_heap || ua->timeout_heap->size == 0)) ? NULL : &tv; - - int nfds = select(ua->max_fd + 1, &readfds, &writefds, &exceptfds, ptv); - if (nfds < 0) { - if (errno == EINTR) return; - perror("select"); + /* Build pollfd array from socket list */ + int socket_count = 0; + struct socket_node* cur = ua->socket_head; + while (cur) { + socket_count++; + cur = cur->next; + } + + int wakeup_fd_present = ua->wakeup_initialized ? 1 : 0; + int total_fds = socket_count + wakeup_fd_present; + + if (total_fds == 0) { + /* No sockets and no wakeup fd, just wait for timeout */ + if (timeout_ms >= 0) { + /* usleep would be better but we just call poll with empty set */ + struct pollfd dummy; + poll(&dummy, 0, timeout_ms); + } else { + /* Infinite timeout with no sockets - should not happen in practice */ + return; + } + /* Check timeouts again after sleep */ + process_timeouts(ua); return; } - - /* Process timeouts that may have expired during select */ - process_timeouts(ua); - - /* Process sockets with faster dispatch */ - for (int fd = 0; nfds > 0 && fd <= ua->max_fd; fd++) { - struct socket_node* node = ua->fd_to_node[fd]; - if (!node) continue; - - if (node->except_cbk && FD_ISSET(fd, &exceptfds)) { - node->except_cbk(fd, node->user_data); - nfds--; + + struct pollfd* fds = malloc(total_fds * sizeof(struct pollfd)); + struct socket_node** nodes = NULL; + if (socket_count > 0) { + nodes = malloc(socket_count * sizeof(struct socket_node*)); + } + if (!fds || (socket_count > 0 && !nodes)) { + free(fds); + free(nodes); + return; /* out of memory */ + } + + /* Fill arrays */ + int idx = 0; + + /* Add wakeup fd first if present */ + if (wakeup_fd_present) { + fds[idx].fd = ua->wakeup_pipe[0]; + fds[idx].events = POLLIN; + fds[idx].revents = 0; + idx++; + } + + /* Add socket fds */ + cur = ua->socket_head; + while (cur) { + fds[idx].fd = cur->fd; + fds[idx].events = 0; + fds[idx].revents = 0; + + if (cur->read_cbk) fds[idx].events |= POLLIN; + if (cur->write_cbk) fds[idx].events |= POLLOUT; + if (cur->except_cbk) fds[idx].events |= POLLPRI; + + if (nodes) { + nodes[idx - wakeup_fd_present] = cur; } - if (node->read_cbk && FD_ISSET(fd, &readfds)) { - node->read_cbk(fd, node->user_data); - nfds--; + idx++; + cur = cur->next; + } + + /* Call poll */ + int ret = poll(fds, total_fds, timeout_ms); + if (ret < 0) { + if (errno == EINTR) { + free(fds); + free(nodes); + return; } - if (node->write_cbk && FD_ISSET(fd, &writefds)) { - node->write_cbk(fd, node->user_data); - nfds--; + perror("poll"); + free(fds); + free(nodes); + return; + } + + /* Process timeouts that may have expired during poll */ + process_timeouts(ua); + + /* Process socket events */ + if (ret > 0) { + for (int i = 0; i < total_fds; i++) { + if (fds[i].revents == 0) continue; + + /* Handle wakeup fd separately */ + if (wakeup_fd_present && i == 0) { + if (fds[i].revents & POLLIN) { + drain_wakeup_pipe(ua); + } + continue; + } + + /* Socket event */ + int socket_idx = i - wakeup_fd_present; + struct socket_node* node = nodes[socket_idx]; + + /* Check for error conditions first */ + if (fds[i].revents & (POLLERR | POLLHUP | POLLNVAL)) { + /* Treat as exceptional condition */ + if (node->except_cbk) { + node->except_cbk(node->fd, node->user_data); + } + } + + /* Exceptional data (out-of-band) */ + if (fds[i].revents & POLLPRI) { + if (node->except_cbk) { + node->except_cbk(node->fd, node->user_data); + } + } + + /* Read readiness */ + if (fds[i].revents & POLLIN) { + if (node->read_cbk) { + node->read_cbk(node->fd, node->user_data); + } + } + + /* Write readiness */ + if (fds[i].revents & POLLOUT) { + if (node->write_cbk) { + node->write_cbk(node->fd, node->user_data); + } + } } } + + free(fds); + free(nodes); } @@ -340,14 +433,31 @@ uasync_t* uasync_create(void) { if (!ua) return NULL; memset(ua, 0, sizeof(struct uasync_s)); - ua->max_fd = -1; - FD_ZERO(&ua->master_readfds); - FD_ZERO(&ua->master_writefds); - FD_ZERO(&ua->master_exceptfds); - memset(ua->fd_to_node, 0, sizeof(ua->fd_to_node)); + ua->wakeup_pipe[0] = -1; + ua->wakeup_pipe[1] = -1; + ua->wakeup_initialized = 0; + + // Create wakeup pipe + if (pipe(ua->wakeup_pipe) < 0) { + fprintf(stderr, "[UASYNC WARN] Failed to create wakeup pipe: %s\n", strerror(errno)); + // Continue without wakeup mechanism + ua->wakeup_pipe[0] = -1; + ua->wakeup_pipe[1] = -1; + } else { + ua->wakeup_initialized = 1; + // Set non-blocking on read end to avoid blocking if pipe is full + int flags = fcntl(ua->wakeup_pipe[0], F_GETFL, 0); + if (flags >= 0) { + fcntl(ua->wakeup_pipe[0], F_SETFL, flags | O_NONBLOCK); + } + } ua->timeout_heap = timeout_heap_create(16); if (!ua->timeout_heap) { + if (ua->wakeup_initialized) { + close(ua->wakeup_pipe[0]); + close(ua->wakeup_pipe[1]); + } free(ua); return NULL; } @@ -393,6 +503,12 @@ void uasync_destroy(uasync_t* ua) { cur = next; } + // Close wakeup pipe + if (ua->wakeup_initialized) { + close(ua->wakeup_pipe[0]); + close(ua->wakeup_pipe[1]); + } + // Final leak check if (ua->timer_alloc_count != ua->timer_free_count || ua->socket_alloc_count != ua->socket_free_count) { fprintf(stderr, "[UASYNC FATAL] Memory leaks detected after cleanup: timers %zu/%zu, sockets %zu/%zu\n", @@ -406,12 +522,6 @@ void uasync_destroy(uasync_t* ua) { void uasync_init_instance(uasync_t* ua) { if (!ua) return; - ua->max_fd = -1; - FD_ZERO(&ua->master_readfds); - FD_ZERO(&ua->master_writefds); - FD_ZERO(&ua->master_exceptfds); - memset(ua->fd_to_node, 0, sizeof(ua->fd_to_node)); - if (!ua->timeout_heap) { ua->timeout_heap = timeout_heap_create(16); if (ua->timeout_heap) { @@ -431,3 +541,21 @@ void uasync_get_stats(uasync_t* ua, size_t* timer_alloc, size_t* timer_free, siz // Get global instance for backward compatibility +// Wakeup mechanism +int uasync_wakeup(uasync_t* ua) { + if (!ua || !ua->wakeup_initialized) return -1; + + char byte = 0; + ssize_t ret = write(ua->wakeup_pipe[1], &byte, 1); + if (ret != 1) { + // Don't print error from signal handler + return -1; + } + return 0; +} + +int uasync_get_wakeup_fd(uasync_t* ua) { + if (!ua || !ua->wakeup_initialized) return -1; + return ua->wakeup_pipe[1]; +} + diff --git a/u_async/u_async.h b/u_async/u_async.h index 18ab6287..f760375d 100644 --- a/u_async/u_async.h +++ b/u_async/u_async.h @@ -6,7 +6,6 @@ #define UASYNC_H #include -#include #include typedef void (*timeout_callback_t)(void* user_arg);// передаёт user_arg из uasync_set_timeout @@ -44,4 +43,8 @@ void uasync_mainloop(uasync_t* ua); // Debug statistics void uasync_get_stats(uasync_t* ua, size_t* timer_alloc, size_t* timer_free, size_t* socket_alloc, size_t* socket_free); +// Wakeup mechanism for interrupting poll +int uasync_wakeup(uasync_t* ua); +int uasync_get_wakeup_fd(uasync_t* ua); // returns write fd for wakeup pipe (for signal handlers) + #endif // UASYNC_H