jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Оптимизация pacing механизма и устранение race condition 

- Устранена критическая race condition: callback больше не пытается отменить уже удаленный waiter
- Упрощена логика управления waiter'ами: callback просто очищает указатель и отправляет следующий пакет
- Добавлена статистика pacing для анализа производительности
- Оптимизирована производительность: убраны лишние отладочные сообщения в hot path
- Улучшена обработка NULL от queue_wait_threshold: разделены случаи очередь пуста и ошибка
- Добавлена условная компиляция для отладочных сообщений в ll_queue.c
- Pacing теперь работает быстро при нулевых очередях: callback вызывается немедленно при опустошении
v2_dev
Evgeny 9 months ago
parent
e3c94a4004
commit
7339b76fb8
10 changed files with 809 additions and 511 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 7
      AGENTS.md
  2. 10
      changelog.txt
  3. 560
      src/config_parser.c
  4. 102
      src/config_parser.h
  5. 10
      src/control_socket.c
  6. 12
      src/ll_queue.c
  7. 2
      src/ll_queue.h
  8. 320
      src/utun.c
  9. 292
      u_async/u_async.c
  10. 5
      u_async/u_async.h

7
AGENTS.md
Unescape View File

@ -6,10 +6,15 @@ This document provides essential information for agentic coding assistants worki
Если не работает - добавляй отладочную информацию чтобы быстрее найти проблемное место и не рушить работающий код.
если отладочная информация будет флудить сделай ее отключаемой или подумай как ограничить ее вывод по возможности сохраняя информативность.
Добавляй в код структуры для статистики (например суммируй число ошибок, вызовов и других потенциально нужных для анализа при ошибках метрик).
добавляй в тест подсчет времени сколько каждфй этап длился для лучшего понимания оптимизации
добавляй в тест подсчет времени сколько каждый этап длился для лучшего понимания оптимизации
перед запуском тестов не забывай их пересобирать
веди changelog.txt: дата время: что поменялось
задача протокола - обеспечивать минимальную задержку при большом трафике. т.е. держать необходимый минимум данных в очередях.
дизайн - однопоточный асинхронный.
все ожидающие операции - через сокеты(дескрипторы) и таймауты -> u_async
все очереди через ll_queue
главный модуль обслуживания подключения - connection
## Code Style Guidelines

10
changelog.txt
Unescape View File

@ -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 вызывается немедленно при опустошении

560
src/config_parser.c
Unescape View File

@ -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, &current_conn) != 0) {
fprintf(stderr, "Failed to add connection: %s\n", current_conn.name);
}
}
memset(&current_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, &current_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, &current_server) != 0) {
fprintf(stderr, "Failed to add server: %s\n", current_server.name);
}
}
memset(&current_server, 0, sizeof(current_server));
in_server_section = 0;
}
if (in_client_section) {
if (strlen(current_client.name) > 0) {
if (add_client(config, &current_client) != 0) {
fprintf(stderr, "Failed to add client: %s\n", current_client.name);
}
}
memset(&current_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, &current_conn_v2) != 0) {
fprintf(stderr, "Failed to add connection v2\n");
}
}
memset(&current_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(&current_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, &current_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, &current_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, &current_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;
}

102
src/config_parser.h
Unescape View File

@ -4,7 +4,6 @@
#include <stdint.h>
#include <stddef.h>
#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

10
src/control_socket.c
Unescape View File

@ -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;

12
src/ll_queue.c
Unescape View File

@ -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;
}

2
src/ll_queue.h
Unescape View File

@ -121,7 +121,7 @@ static inline size_t ll_entry_size(ll_entry_t* entry) {
// ==================== Асинхронное ожидание ====================
// Зарегистрировать коллбэк, который будет вызван когда очередь будет иметь
// Зарегистрировать одноразовый коллбэк, который будет вызван когда очередь будет иметь
// не более max_packets пакетов и не более max_bytes байт.
// Если условие уже выполнено, коллбэк вызывается немедленно.
// Можно зарегистрировать несколько ожиданий на одной очереди.

320
src/utun.c
Unescape View File

@ -18,10 +18,13 @@
#include <sys/stat.h>
#include <errno.h>
#include <signal.h>
#include <poll.h>
#include <getopt.h>
#include <arpa/inet.h>
// 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

292
u_async/u_async.c
Unescape View File

@ -7,10 +7,11 @@
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <poll.h>
#include <limits.h>
#include <fcntl.h>
#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];
}

5
u_async/u_async.h
Unescape View File

@ -6,7 +6,6 @@
#define UASYNC_H
#include <sys/time.h>
#include <sys/select.h>
#include <stddef.h>
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

Write Preview
Loading…
Cancel
Save