Browse Source
lib/u_async: uasync_set_socket_read/write — O(1) EPOLL_CTL_MOD toggle per socket lib/ll_queue: queue_set_empty_callback — one-shot deferred callback on queue drain lib/tcp_io: TCP connection management via uasync + ll_queue with pool alloc - read: direct recv into pool buffer, high/low water flow control - write: autofetch via ll_queue callback, partial send via write_buf - on_fin/on_error/on_flushed callbacks, FIN deferred until read_queue drained - getpeername for pre-connected socket detection tcp_proxy_server: migrated to tcp_io (22→13 fields, 437→270 lines, -7 functions) tests: test_tcp_io (5 tests), test_tcp_proxy_server rewritten for tcp_connetcp-inflight-fix
Evgeny
4 months ago
14 changed files with 1216 additions and 422 deletions
@ -0,0 +1,358 @@
|
||||
// tcp_io.c — управление TCP-соединением через uasync + ll_queue
|
||||
#include "tcp_io.h" |
||||
#include "debug_config.h" |
||||
#include "mem.h" |
||||
#include <stdlib.h> |
||||
#include <string.h> |
||||
#include <errno.h> |
||||
#ifndef _WIN32 |
||||
#include <unistd.h> |
||||
#include <sys/socket.h> |
||||
#include <fcntl.h> |
||||
#endif |
||||
#ifndef MSG_NOSIGNAL |
||||
#define MSG_NOSIGNAL 0 |
||||
#endif |
||||
|
||||
static void read_cb(socket_t sock, void* arg); |
||||
static void write_cb(socket_t sock, void* arg); |
||||
static void error_cb(socket_t sock, void* arg); |
||||
static void resume_read_cb(struct ll_queue* q, void* arg); |
||||
static void fin_deferred_cb(struct ll_queue* q, void* arg); |
||||
static void write_queue_fetch_cb(struct ll_queue* q, void* arg); |
||||
static void flush_write_buf(struct tcp_conn* tc); |
||||
|
||||
struct tcp_conn* tcp_conn_create( |
||||
struct UASYNC* ua, socket_t sock, |
||||
size_t entry_data_size, size_t write_chunk_size, |
||||
int read_high_water, int read_low_water, |
||||
void (*on_fin)(struct tcp_conn* tc, void* arg), |
||||
void (*on_error)(struct tcp_conn* tc, int err, void* arg), |
||||
void* arg) |
||||
{ |
||||
if (!ua || sock == SOCKET_INVALID) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: invalid args ua=%p", ua); return NULL; } |
||||
|
||||
struct tcp_conn* tc = u_calloc(1, sizeof(struct tcp_conn)); |
||||
if (!tc) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: u_calloc failed"); return NULL; } |
||||
|
||||
tc->ua = ua; |
||||
tc->sock = sock; |
||||
tc->entry_data_size = entry_data_size; |
||||
tc->write_chunk_size = write_chunk_size; |
||||
tc->read_high_water = read_high_water; |
||||
tc->read_low_water = read_low_water; |
||||
tc->on_fin = on_fin; |
||||
tc->on_error = on_error; |
||||
tc->arg = arg; |
||||
|
||||
tc->entry_pool = memory_pool_init(sizeof(struct ll_entry)); |
||||
{ |
||||
size_t ds = entry_data_size > write_chunk_size ? entry_data_size : write_chunk_size; |
||||
tc->data_pool = memory_pool_init(ds); |
||||
} |
||||
if (!tc->entry_pool || !tc->data_pool) { |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: memory_pool_init failed"); |
||||
if (tc->entry_pool) memory_pool_destroy(tc->entry_pool); |
||||
if (tc->data_pool) memory_pool_destroy(tc->data_pool); |
||||
u_free(tc); return NULL; |
||||
} |
||||
|
||||
tc->read_queue = queue_new(ua, 0, 0, 0, "tcp_io_rq"); |
||||
tc->write_queue = queue_new(ua, 0, 0, 0, "tcp_io_wq"); |
||||
if (!tc->read_queue || !tc->write_queue) { |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: queue_new failed"); |
||||
if (tc->read_queue) queue_free(tc->read_queue); |
||||
if (tc->write_queue) queue_free(tc->write_queue); |
||||
memory_pool_destroy(tc->entry_pool); |
||||
memory_pool_destroy(tc->data_pool); |
||||
u_free(tc); return NULL; |
||||
} |
||||
queue_set_threshold(tc->read_queue, read_low_water, 0); |
||||
queue_set_callback(tc->write_queue, write_queue_fetch_cb, tc); |
||||
queue_set_waiter_defer(tc->write_queue, 1); |
||||
|
||||
tc->socket_id = uasync_add_socket_t(ua, sock, read_cb, write_cb, error_cb, tc); |
||||
if (!tc->socket_id) { |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: uasync_add_socket_t failed"); |
||||
queue_free(tc->read_queue); |
||||
queue_free(tc->write_queue); |
||||
memory_pool_destroy(tc->entry_pool); |
||||
memory_pool_destroy(tc->data_pool); |
||||
u_free(tc); return NULL; |
||||
} |
||||
tc->write_monitor = 1; |
||||
|
||||
{ |
||||
struct sockaddr_storage addr; |
||||
socklen_t alen = sizeof(addr); |
||||
if (getpeername(sock, (struct sockaddr*)&addr, &alen) == 0) tc->connected = 1; |
||||
} |
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "tcp_conn_create: fd=%d entry=%zu chunk=%zu hw=%d lw=%d connected=%d", |
||||
(int)sock, entry_data_size, write_chunk_size, read_high_water, read_low_water, tc->connected); |
||||
return tc; |
||||
} |
||||
|
||||
void tcp_conn_destroy(struct tcp_conn* tc) { |
||||
if (!tc) return; |
||||
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "tcp_conn_destroy: fd=%d connected=%d error=%d fin=%d", |
||||
(int)tc->sock, tc->connected, tc->error, tc->fin); |
||||
|
||||
if (tc->socket_id) { |
||||
uasync_remove_socket_t(tc->ua, tc->sock); |
||||
tc->socket_id = NULL; |
||||
} |
||||
queue_waiter_cancel(tc->read_queue, &tc->read_waiter); |
||||
queue_set_empty_callback(tc->read_queue, NULL, NULL); |
||||
|
||||
struct ll_entry* e; |
||||
while ((e = queue_data_get(tc->read_queue)) != NULL) { |
||||
if (e->dgram) memory_pool_free(tc->data_pool, e->dgram); |
||||
queue_entry_free(e); queue_resume_callback(tc->read_queue); |
||||
} |
||||
while ((e = queue_data_get(tc->write_queue)) != NULL) { |
||||
if (e->dgram) memory_pool_free(tc->data_pool, e->dgram); |
||||
queue_entry_free(e); queue_resume_callback(tc->write_queue); |
||||
} |
||||
queue_free(tc->read_queue); |
||||
queue_free(tc->write_queue); |
||||
|
||||
if (tc->write_buf) memory_pool_free(tc->data_pool, tc->write_buf); |
||||
memory_pool_destroy(tc->entry_pool); |
||||
memory_pool_destroy(tc->data_pool); |
||||
u_free(tc); |
||||
} |
||||
|
||||
// ====================================================================
|
||||
// Чтение из сокета
|
||||
// ====================================================================
|
||||
|
||||
static void read_cb(socket_t sock, void* arg) { |
||||
(void)sock; |
||||
struct tcp_conn* tc = (struct tcp_conn*)arg; |
||||
if (!tc || tc->sock == SOCKET_INVALID) return; |
||||
|
||||
struct ll_entry* e = queue_entry_new_from_pool(tc->entry_pool); |
||||
uint8_t* buf = memory_pool_alloc(tc->data_pool); |
||||
if (!e || !buf) { |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: read_cb alloc failed fd=%d", (int)tc->sock); |
||||
if (e) queue_entry_free(e); |
||||
if (buf) memory_pool_free(tc->data_pool, buf); |
||||
uasync_set_socket_read(tc->ua, tc->socket_id, 0); |
||||
tc->read_paused = 1; |
||||
return; |
||||
} |
||||
|
||||
ssize_t n = recv(tc->sock, buf, tc->entry_data_size, 0); |
||||
if (n > 0) { |
||||
e->dgram = buf; |
||||
e->len = (uint16_t)n; |
||||
queue_data_put(tc->read_queue, e); |
||||
if (tc->read_queue->count >= tc->read_high_water && !tc->read_paused) { |
||||
uasync_set_socket_read(tc->ua, tc->socket_id, 0); |
||||
tc->read_paused = 1; |
||||
queue_waiter_wait(tc->read_queue, &tc->read_waiter, resume_read_cb, tc); |
||||
} |
||||
} else if (n == 0) { |
||||
memory_pool_free(tc->data_pool, buf); |
||||
queue_entry_free(e); |
||||
tc->fin = 1; |
||||
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "tcp_io: FIN fd=%d", (int)tc->sock); |
||||
uasync_set_socket_read(tc->ua, tc->socket_id, 0); |
||||
if (tc->read_queue->count == 0) { |
||||
if (tc->on_fin) tc->on_fin(tc, tc->arg); |
||||
} else { |
||||
queue_set_empty_callback(tc->read_queue, fin_deferred_cb, tc); |
||||
} |
||||
} else { |
||||
memory_pool_free(tc->data_pool, buf); |
||||
queue_entry_free(e); |
||||
if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) return; |
||||
tc->error = 1; |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: recv error fd=%d errno=%d", (int)tc->sock, errno); |
||||
if (tc->on_error) tc->on_error(tc, errno, tc->arg); |
||||
} |
||||
} |
||||
|
||||
static void resume_read_cb(struct ll_queue* q, void* arg) { |
||||
(void)q; |
||||
struct tcp_conn* tc = (struct tcp_conn*)arg; |
||||
if (!tc || tc->sock == SOCKET_INVALID) return; |
||||
uasync_set_socket_read(tc->ua, tc->socket_id, 1); |
||||
tc->read_paused = 0; |
||||
} |
||||
|
||||
static void fin_deferred_cb(struct ll_queue* q, void* arg) { |
||||
(void)q; |
||||
struct tcp_conn* tc = (struct tcp_conn*)arg; |
||||
if (tc->on_fin) tc->on_fin(tc, tc->arg); |
||||
} |
||||
|
||||
// ====================================================================
|
||||
// Запись в сокет
|
||||
// ====================================================================
|
||||
|
||||
static void flush_write_buf(struct tcp_conn* tc) { |
||||
while (tc->write_buf && tc->write_offset < tc->write_len) { |
||||
ssize_t n = send(tc->sock, tc->write_buf + tc->write_offset, tc->write_len - tc->write_offset, MSG_NOSIGNAL); |
||||
if (n > 0) { |
||||
tc->write_offset += (size_t)n; |
||||
if (tc->write_offset >= tc->write_len) { |
||||
memory_pool_free(tc->data_pool, tc->write_buf); |
||||
tc->write_buf = NULL; |
||||
tc->write_len = 0; |
||||
tc->write_offset = 0; |
||||
return; |
||||
} |
||||
continue; |
||||
} |
||||
if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) return; |
||||
tc->error = 1; |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: send error fd=%d errno=%d", (int)tc->sock, errno); |
||||
if (tc->on_error) tc->on_error(tc, errno, tc->arg); |
||||
return; |
||||
} |
||||
} |
||||
|
||||
static void write_queue_fetch_cb(struct ll_queue* q, void* arg) { |
||||
struct tcp_conn* tc = (struct tcp_conn*)arg; |
||||
|
||||
if (tc->error || tc->fin) return; |
||||
|
||||
if (tc->write_buf) { |
||||
flush_write_buf(tc); |
||||
if (tc->write_buf) return; // остался остаток, ждём write_cb
|
||||
} |
||||
|
||||
if (!tc->connected) return; // ждём connect
|
||||
|
||||
struct ll_entry* e = queue_data_get(q); |
||||
if (!e) { |
||||
uasync_set_socket_write(tc->ua, tc->socket_id, 0); |
||||
tc->write_monitor = 0; |
||||
queue_resume_callback(q); |
||||
if (tc->on_flushed) { |
||||
void (*cb)(struct tcp_conn*, void*) = tc->on_flushed; |
||||
tc->on_flushed = NULL; |
||||
cb(tc, tc->arg); |
||||
} |
||||
return; |
||||
} |
||||
|
||||
ssize_t n = send(tc->sock, e->dgram, e->len, MSG_NOSIGNAL); |
||||
if (n == (ssize_t)e->len) { |
||||
memory_pool_free(tc->data_pool, e->dgram); |
||||
queue_entry_free(e); |
||||
queue_resume_callback(q); |
||||
} else if (n > 0) { |
||||
size_t rem = e->len - (size_t)n; |
||||
tc->write_buf = memory_pool_alloc(tc->data_pool); |
||||
if (!tc->write_buf) { |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: write_buf alloc failed fd=%d", (int)tc->sock); |
||||
tc->error = 1; |
||||
memory_pool_free(tc->data_pool, e->dgram); |
||||
queue_entry_free(e); |
||||
return; |
||||
} |
||||
memcpy(tc->write_buf, e->dgram + n, rem); |
||||
tc->write_len = rem; |
||||
tc->write_offset = 0; |
||||
memory_pool_free(tc->data_pool, e->dgram); |
||||
queue_entry_free(e); |
||||
uasync_set_socket_write(tc->ua, tc->socket_id, 1); |
||||
tc->write_monitor = 1; |
||||
} else if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) { |
||||
tc->write_buf = memory_pool_alloc(tc->data_pool); |
||||
if (!tc->write_buf) { |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: write_buf alloc failed fd=%d", (int)tc->sock); |
||||
tc->error = 1; |
||||
memory_pool_free(tc->data_pool, e->dgram); |
||||
queue_entry_free(e); |
||||
return; |
||||
} |
||||
memcpy(tc->write_buf, e->dgram, e->len); |
||||
tc->write_len = e->len; |
||||
tc->write_offset = 0; |
||||
memory_pool_free(tc->data_pool, e->dgram); |
||||
queue_entry_free(e); |
||||
uasync_set_socket_write(tc->ua, tc->socket_id, 1); |
||||
tc->write_monitor = 1; |
||||
} else { |
||||
tc->error = 1; |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: send error fd=%d errno=%d", (int)tc->sock, errno); |
||||
if (tc->on_error) tc->on_error(tc, errno, tc->arg); |
||||
memory_pool_free(tc->data_pool, e->dgram); |
||||
queue_entry_free(e); |
||||
} |
||||
} |
||||
|
||||
static void write_cb(socket_t sock, void* arg) { |
||||
(void)sock; |
||||
struct tcp_conn* tc = (struct tcp_conn*)arg; |
||||
if (!tc || tc->sock == SOCKET_INVALID) return; |
||||
|
||||
if (!tc->connected) { |
||||
int err = 0; |
||||
socklen_t len = sizeof(err); |
||||
if (getsockopt(tc->sock, SOL_SOCKET, SO_ERROR, &err, &len) == 0 && err == 0) { |
||||
tc->connected = 1; |
||||
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "tcp_io: connect ok fd=%d", (int)tc->sock); |
||||
} else { |
||||
tc->error = 1; |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: connect fail fd=%d err=%d", (int)tc->sock, err); |
||||
if (tc->on_error) tc->on_error(tc, err ? err : -1, tc->arg); |
||||
return; |
||||
} |
||||
} |
||||
|
||||
flush_write_buf(tc); |
||||
if (!tc->write_buf) queue_resume_callback(tc->write_queue); |
||||
} |
||||
|
||||
// ====================================================================
|
||||
// Асинхронная ошибка сокета
|
||||
// ====================================================================
|
||||
|
||||
static void error_cb(socket_t sock, void* arg) { |
||||
(void)sock; |
||||
struct tcp_conn* tc = (struct tcp_conn*)arg; |
||||
if (!tc || tc->sock == SOCKET_INVALID) return; |
||||
tc->error = 1; |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: async error fd=%d", (int)tc->sock); |
||||
if (tc->on_error) tc->on_error(tc, -1, tc->arg); |
||||
} |
||||
|
||||
void tcp_conn_set_flushed(struct tcp_conn* tc, void (*on_flushed)(struct tcp_conn* tc, void* arg)) { |
||||
if (!tc) return; |
||||
tc->on_flushed = on_flushed; |
||||
} |
||||
|
||||
// ====================================================================
|
||||
// Отправка данных в сокет (внешний интерфейс)
|
||||
// ====================================================================
|
||||
|
||||
int tcp_conn_push_write(struct tcp_conn* tc, const uint8_t* data, size_t len) { |
||||
if (!tc || !data || len == 0) return -1; |
||||
if (tc->error || tc->fin) return -1; |
||||
|
||||
size_t offset = 0; |
||||
while (offset < len) { |
||||
size_t chunk = len - offset; |
||||
if (chunk > tc->write_chunk_size) chunk = tc->write_chunk_size; |
||||
|
||||
struct ll_entry* e = queue_entry_new_from_pool(tc->entry_pool); |
||||
uint8_t* buf = memory_pool_alloc(tc->data_pool); |
||||
if (!e || !buf) { |
||||
DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_io: push_write alloc failed fd=%d chunk=%zu", (int)tc->sock, chunk); |
||||
if (e) queue_entry_free(e); |
||||
if (buf) memory_pool_free(tc->data_pool, buf); |
||||
return -1; |
||||
} |
||||
memcpy(buf, data + offset, chunk); |
||||
e->dgram = buf; |
||||
e->len = (uint16_t)chunk; |
||||
queue_data_put(tc->write_queue, e); |
||||
offset += chunk; |
||||
} |
||||
return 0; |
||||
} |
||||
@ -0,0 +1,96 @@
|
||||
// tcp_io.h — управление TCP-соединением через uasync + ll_queue
|
||||
//
|
||||
// Один tcp_conn = одно TCP-соединение. Две очереди: read_queue (сокет → данные) и
|
||||
// write_queue (данные → сокет). Обе работают через автозабор ll_queue (deferred).
|
||||
//
|
||||
// Пул-аллокация:
|
||||
// entry_pool — только struct ll_entry (без inline data)
|
||||
// data_pool — буферы данных чтения/записи (max(entry_data_size, write_chunk_size))
|
||||
// Данные и структуры аллоцируются раздельно — не копируются при recv/send.
|
||||
//
|
||||
// Чтение:
|
||||
// read_cb → memory_pool_alloc(read_pool) → recv() прямо в буфер → entry в read_queue
|
||||
// read_queue автозабор(deferred) → внешний коллбэк (например, отправка в ETCP)
|
||||
// high_water → пауза чтения (убираем EPOLLIN), low_water/waiter → возобновление
|
||||
// FIN: если read_queue не пуст → empty_callback откладывает on_fin; иначе сразу
|
||||
//
|
||||
// Запись:
|
||||
// Внешний код: tcp_conn_push_write(data, len) → аллокация через write_pool → entry в write_queue
|
||||
// write_queue автозабор(deferred): fetch_cb → send() в сокет
|
||||
// EAGAIN → write_buf (из data_pool) + EPOLLOUT ON → write_cb досылает → resume автозабора
|
||||
// Всё отправлено → EPOLLOUT OFF + on_flushed (если установлен)
|
||||
//
|
||||
// Connect:
|
||||
// tcp_conn_create регистрирует сокет с read_cb + write_cb в uasync
|
||||
// После create вызывается connect() (неблокирующий, EINPROGRESS)
|
||||
// write_cb детектит завершение connect через getsockopt(SO_ERROR)
|
||||
// getpeername() сразу после create проверяет pre-connected сокеты (socketpair)
|
||||
// До connect данные копятся в write_queue (автозабор не пытается send на unconnected сокет)
|
||||
|
||||
#ifndef TCP_IO_H |
||||
#define TCP_IO_H |
||||
|
||||
#include "u_async.h" |
||||
#include "socket_compat.h" |
||||
#include "ll_queue.h" |
||||
#include "memory_pool.h" |
||||
|
||||
struct tcp_conn { |
||||
socket_t sock; |
||||
struct UASYNC* ua; |
||||
void* socket_id; |
||||
|
||||
struct ll_queue* read_queue; // сокет → данные (блоки до entry_data_size)
|
||||
struct ll_queue* write_queue; // данные → сокет (блоки до write_chunk_size)
|
||||
|
||||
int read_high_water; |
||||
int read_low_water; |
||||
uint8_t read_paused; |
||||
|
||||
uint8_t write_monitor; // 1 = EPOLLOUT активен
|
||||
uint8_t connected; |
||||
uint8_t error; |
||||
uint8_t fin; // FIN от сокета
|
||||
|
||||
// Частичная отправка (из write_pool, не в очереди — досылается первой)
|
||||
uint8_t* write_buf; |
||||
size_t write_len; |
||||
size_t write_offset; |
||||
size_t write_chunk_size; |
||||
|
||||
// Пулы памяти
|
||||
struct memory_pool* entry_pool; // sizeof(struct ll_entry)
|
||||
struct memory_pool* data_pool; // max(entry_data_size, write_chunk_size)
|
||||
size_t entry_data_size; |
||||
|
||||
struct queue_waiter_handle read_waiter; |
||||
|
||||
// Коллбэки
|
||||
void (*on_fin)(struct tcp_conn* tc, void* arg); |
||||
void (*on_error)(struct tcp_conn* tc, int err, void* arg); |
||||
void (*on_flushed)(struct tcp_conn* tc, void* arg); // все данные записи отправлены
|
||||
void* arg; |
||||
}; |
||||
|
||||
struct tcp_conn* tcp_conn_create( |
||||
struct UASYNC* ua, |
||||
socket_t sock, |
||||
size_t entry_data_size, |
||||
size_t write_chunk_size, |
||||
int read_high_water, |
||||
int read_low_water, |
||||
void (*on_fin)(struct tcp_conn* tc, void* arg), |
||||
void (*on_error)(struct tcp_conn* tc, int err, void* arg), |
||||
void* arg); |
||||
|
||||
void tcp_conn_destroy(struct tcp_conn* tc); |
||||
|
||||
// Отправка данных в сокет: режет на чанки по write_chunk_size, кладёт в write_queue.
|
||||
// Автозабор write_queue сам отправит когда сокет будет готов.
|
||||
int tcp_conn_push_write(struct tcp_conn* tc, const uint8_t* data, size_t len); |
||||
|
||||
// Одноразовый коллбэк: вызывается когда write_queue + write_buf полностью опустели.
|
||||
// После вызова сбрасывается. Установить повторно можно в любой момент.
|
||||
void tcp_conn_set_flushed(struct tcp_conn* tc, void (*on_flushed)(struct tcp_conn* tc, void* arg)); |
||||
|
||||
#endif |
||||
@ -0,0 +1,298 @@
|
||||
// test_tcp_io.c — тесты библиотеки tcp_conn
|
||||
#include <stdio.h> |
||||
#include <stdlib.h> |
||||
#include <string.h> |
||||
#include <assert.h> |
||||
#include <errno.h> |
||||
#include "../lib/platform_compat.h" |
||||
#include "../lib/u_async.h" |
||||
#include "../lib/tcp_io.h" |
||||
#include "../lib/debug_config.h" |
||||
#include "../lib/mem.h" |
||||
|
||||
#ifndef MSG_NOSIGNAL |
||||
#define MSG_NOSIGNAL 0 |
||||
#endif |
||||
|
||||
static int tests_run = 0; |
||||
static int tests_passed = 0; |
||||
static int tests_failed = 0; |
||||
|
||||
#define TEST_START(name) do { DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "TEST: %s... ", name); tests_run++; } while(0) |
||||
#define TEST_PASS() do { DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "PASS"); tests_passed++; } while(0) |
||||
#define TEST_FAIL(msg) do { DEBUG_ERROR(DEBUG_CATEGORY_UASYNC, "FAIL: %s", msg); tests_failed++; return; } while(0) |
||||
#define ASSERT_TRUE(cond, msg) do { if (!(cond)) TEST_FAIL(msg); } while(0) |
||||
#define ASSERT_EQ(a, b, msg) do { if ((a) != (b)) TEST_FAIL(msg); } while(0) |
||||
|
||||
static int g_fin_count = 0; |
||||
static int g_error_count = 0; |
||||
static int g_last_error = 0; |
||||
static struct tcp_conn* g_last_fin_tc = NULL; |
||||
|
||||
static void on_fin(struct tcp_conn* tc, void* arg) { |
||||
(void)arg; |
||||
g_fin_count++; |
||||
g_last_fin_tc = tc; |
||||
} |
||||
|
||||
static void on_error(struct tcp_conn* tc, int err, void* arg) { |
||||
(void)arg; |
||||
g_error_count++; |
||||
g_last_error = err; |
||||
} |
||||
|
||||
static void reset_counters(void) { |
||||
g_fin_count = 0; |
||||
g_error_count = 0; |
||||
g_last_error = 0; |
||||
g_last_fin_tc = NULL; |
||||
} |
||||
|
||||
static void test_basic_send_recv(void) { |
||||
TEST_START("Basic send and recv via tcp_conn"); |
||||
|
||||
uasync_t* ua = uasync_create(); |
||||
ASSERT_TRUE(ua != NULL, "uasync_create failed"); |
||||
reset_counters(); |
||||
|
||||
int sv[2]; |
||||
ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0, "socketpair failed"); |
||||
for (int i = 0; i < 2; i++) fcntl(sv[i], F_SETFL, fcntl(sv[i], F_GETFL, 0) | O_NONBLOCK); |
||||
|
||||
struct tcp_conn* tc = tcp_conn_create(ua, sv[0], 1500, 8192, 32, 8, on_fin, on_error, NULL); |
||||
ASSERT_TRUE(tc != NULL, "tcp_conn_create failed"); |
||||
|
||||
// Отправляем данные через tcp_conn
|
||||
uint8_t send_buf[3000]; |
||||
memset(send_buf, 'A', sizeof(send_buf)); |
||||
int ret = tcp_conn_push_write(tc, send_buf, sizeof(send_buf)); |
||||
ASSERT_EQ(ret, 0, "tcp_conn_push_write first call failed"); |
||||
uasync_poll(ua, 1); // write_cb должен отправить
|
||||
|
||||
// Читаем с другой стороны
|
||||
uint8_t recv_buf[4096] = {0}; |
||||
ssize_t total = 0; |
||||
while (total < (ssize_t)sizeof(send_buf)) { |
||||
ssize_t n = recv(sv[1], recv_buf + total, sizeof(recv_buf) - total, 0); |
||||
ASSERT_TRUE(n >= 0, "recv failed on peer socket"); |
||||
total += n; |
||||
} |
||||
ASSERT_TRUE(memcmp(send_buf, recv_buf, sizeof(send_buf)) == 0, "received data mismatch"); |
||||
|
||||
// Отправляем с другой стороны — должно появиться в read_queue
|
||||
uint8_t peer_data[500]; |
||||
memset(peer_data, 'B', sizeof(peer_data)); |
||||
ssize_t wret = send(sv[1], peer_data, sizeof(peer_data), MSG_NOSIGNAL); |
||||
ASSERT_TRUE(wret == sizeof(peer_data), "send on peer socket failed"); |
||||
uasync_poll(ua, 10); |
||||
|
||||
struct ll_entry* e = queue_data_get(tc->read_queue); |
||||
ASSERT_TRUE(e != NULL, "no data in read_queue"); |
||||
ASSERT_EQ(e->len, (int)sizeof(peer_data), "read_queue entry length mismatch"); |
||||
ASSERT_TRUE(memcmp(e->dgram, peer_data, sizeof(peer_data)) == 0, "read_queue data mismatch"); |
||||
queue_entry_free(e); |
||||
queue_resume_callback(tc->read_queue); |
||||
|
||||
// Закрываем сокет — должен вызвать on_fin
|
||||
close(sv[1]); |
||||
uasync_poll(ua, 10); |
||||
ASSERT_EQ(g_fin_count, 1, "on_fin not called"); |
||||
ASSERT_EQ(tc->fin, 1, "tc->fin not set"); |
||||
|
||||
tcp_conn_destroy(tc); |
||||
close(sv[0]); |
||||
uasync_destroy(ua, 0); |
||||
TEST_PASS(); |
||||
} |
||||
|
||||
static void test_partial_write(void) { |
||||
TEST_START("Partial write recovery"); |
||||
|
||||
uasync_t* ua = uasync_create(); |
||||
ASSERT_TRUE(ua != NULL, "uasync_create failed"); |
||||
reset_counters(); |
||||
|
||||
int sv[2]; |
||||
ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0, "socketpair failed"); |
||||
for (int i = 0; i < 2; i++) fcntl(sv[i], F_SETFL, fcntl(sv[i], F_GETFL, 0) | O_NONBLOCK); |
||||
|
||||
struct tcp_conn* tc = tcp_conn_create(ua, sv[0], 1500, 4096, 32, 8, on_fin, on_error, NULL); |
||||
ASSERT_TRUE(tc != NULL, "tcp_conn_create failed"); |
||||
|
||||
// Заполняем приёмный буфер sv[1] маленькими чтениями, чтобы создать EAGAIN на sv[0]
|
||||
// Отправляем много данных быстро, чтобы сокет заполнился
|
||||
int send_size = 256; |
||||
uint8_t* big_buf = u_malloc(200000); |
||||
ASSERT_TRUE(big_buf != NULL, "malloc failed"); |
||||
memset(big_buf, 'X', 200000); |
||||
|
||||
int ret = tcp_conn_push_write(tc, big_buf, 200000); |
||||
ASSERT_EQ(ret, 0, "tcp_conn_push_write large failed"); |
||||
uasync_poll(ua, 1); // write_cb отправляет что может
|
||||
|
||||
// Дрейним sv[1] и проверяем что все данные приходят
|
||||
uint8_t read_buf[8192]; |
||||
ssize_t got = 0; |
||||
int iterations = 0; |
||||
while (got < 200000 && iterations < 1000) { |
||||
uasync_poll(ua, 1); // write_cb может дослать остаток
|
||||
ssize_t n = recv(sv[1], read_buf, sizeof(read_buf), 0); |
||||
if (n > 0) { got += n; iterations = 0; continue; } |
||||
iterations++; |
||||
} |
||||
ASSERT_TRUE(got == 200000, "not all data received"); |
||||
|
||||
u_free(big_buf); |
||||
tcp_conn_destroy(tc); |
||||
close(sv[1]); |
||||
uasync_destroy(ua, 0); |
||||
TEST_PASS(); |
||||
} |
||||
|
||||
static void test_high_water_pause(void) { |
||||
TEST_START("Read high-water pause and resume"); |
||||
|
||||
uasync_t* ua = uasync_create(); |
||||
ASSERT_TRUE(ua != NULL, "uasync_create failed"); |
||||
reset_counters(); |
||||
|
||||
int sv[2]; |
||||
ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0, "socketpair failed"); |
||||
for (int i = 0; i < 2; i++) fcntl(sv[i], F_SETFL, fcntl(sv[i], F_GETFL, 0) | O_NONBLOCK); |
||||
|
||||
// hw=2, lw=0 — пауза после 2 блоков, resume когда пусто
|
||||
struct tcp_conn* tc = tcp_conn_create(ua, sv[0], 64, 8192, 2, 0, on_fin, on_error, NULL); |
||||
ASSERT_TRUE(tc != NULL, "tcp_conn_create failed"); |
||||
|
||||
// Шлём 5 блоков по 64 байта с другой стороны
|
||||
for (int i = 0; i < 5; i++) { |
||||
uint8_t buf[64]; |
||||
memset(buf, (uint8_t)i, sizeof(buf)); |
||||
ssize_t wret = send(sv[1], buf, sizeof(buf), MSG_NOSIGNAL); |
||||
ASSERT_TRUE(wret == sizeof(buf), "send failed on iteration"); |
||||
} |
||||
// Крутим poll пока не упрёмся в high_water (нужно несколько poll-итераций)
|
||||
for (int i = 0; i < 20; i++) { |
||||
uasync_poll(ua, 1); |
||||
if (tc->read_paused) break; |
||||
} |
||||
ASSERT_TRUE(tc->read_paused == 1, "read not paused at high water"); |
||||
|
||||
// Дрейним read_queue — после lw=0 должен сработать resume
|
||||
int drained = 0; |
||||
struct ll_entry* e; |
||||
while ((e = queue_data_get(tc->read_queue)) != NULL) { |
||||
drained++; |
||||
queue_entry_free(e); |
||||
queue_resume_callback(tc->read_queue); |
||||
} |
||||
ASSERT_TRUE(drained >= 2, "should have at least 2 entries"); |
||||
|
||||
uasync_poll(ua, 10); |
||||
ASSERT_TRUE(tc->read_queue->count > 0, "read not resumed"); |
||||
|
||||
// Дрейним остаток
|
||||
while ((e = queue_data_get(tc->read_queue)) != NULL) { |
||||
queue_entry_free(e); |
||||
queue_resume_callback(tc->read_queue); |
||||
} |
||||
|
||||
tcp_conn_destroy(tc); |
||||
close(sv[1]); |
||||
uasync_destroy(ua, 0); |
||||
TEST_PASS(); |
||||
} |
||||
|
||||
static void test_connect_detection(void) { |
||||
TEST_START("Connect completion detection"); |
||||
#ifdef _WIN32 |
||||
TEST_PASS(); |
||||
return; |
||||
#else |
||||
uasync_t* ua = uasync_create(); |
||||
ASSERT_TRUE(ua != NULL, "uasync_create failed"); |
||||
reset_counters(); |
||||
|
||||
int listen_fd = socket(AF_INET, SOCK_STREAM, 0); |
||||
ASSERT_TRUE(listen_fd >= 0, "socket failed"); |
||||
|
||||
struct sockaddr_in addr; |
||||
memset(&addr, 0, sizeof(addr)); |
||||
addr.sin_family = AF_INET; |
||||
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); |
||||
addr.sin_port = 0; |
||||
ASSERT_EQ(bind(listen_fd, (struct sockaddr*)&addr, sizeof(addr)), 0, "bind failed"); |
||||
|
||||
socklen_t alen = sizeof(addr); |
||||
getsockname(listen_fd, (struct sockaddr*)&addr, &alen); |
||||
ASSERT_EQ(listen(listen_fd, 1), 0, "listen failed"); |
||||
|
||||
int client_fd = socket(AF_INET, SOCK_STREAM, 0); |
||||
ASSERT_TRUE(client_fd >= 0, "socket failed"); |
||||
fcntl(client_fd, F_SETFL, fcntl(client_fd, F_GETFL, 0) | O_NONBLOCK); |
||||
|
||||
struct tcp_conn* tc = tcp_conn_create(ua, client_fd, 1500, 8192, 32, 8, on_fin, on_error, NULL); |
||||
ASSERT_TRUE(tc != NULL, "tcp_conn_create failed"); |
||||
ASSERT_EQ(tc->connected, 0, "should not be connected yet"); |
||||
|
||||
int ret = connect(client_fd, (struct sockaddr*)&addr, sizeof(addr)); |
||||
ASSERT_TRUE(ret < 0 && errno == EINPROGRESS, "connect should return EINPROGRESS"); |
||||
|
||||
int server_fd = accept(listen_fd, NULL, NULL); |
||||
ASSERT_TRUE(server_fd >= 0, "accept failed"); |
||||
|
||||
uasync_poll(ua, 100); |
||||
ASSERT_EQ(tc->connected, 1, "connect not detected"); |
||||
|
||||
close(server_fd); |
||||
tcp_conn_destroy(tc); |
||||
close(listen_fd); |
||||
uasync_destroy(ua, 0); |
||||
TEST_PASS(); |
||||
#endif |
||||
} |
||||
|
||||
static void test_error_callback(void) { |
||||
TEST_START("Error callback on broken socket"); |
||||
|
||||
uasync_t* ua = uasync_create(); |
||||
ASSERT_TRUE(ua != NULL, "uasync_create failed"); |
||||
reset_counters(); |
||||
|
||||
int sv[2]; |
||||
ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0, "socketpair failed"); |
||||
for (int i = 0; i < 2; i++) fcntl(sv[i], F_SETFL, fcntl(sv[i], F_GETFL, 0) | O_NONBLOCK); |
||||
|
||||
struct tcp_conn* tc = tcp_conn_create(ua, sv[0], 1500, 8192, 32, 8, on_fin, on_error, NULL); |
||||
ASSERT_TRUE(tc != NULL, "tcp_conn_create failed"); |
||||
|
||||
close(sv[1]); |
||||
uasync_poll(ua, 10); |
||||
ASSERT_EQ(tc->error, 1, "error not set on broken connection"); |
||||
|
||||
tcp_conn_destroy(tc); |
||||
close(sv[0]); |
||||
uasync_destroy(ua, 0); |
||||
TEST_PASS(); |
||||
} |
||||
|
||||
int main(void) { |
||||
debug_config_init(); |
||||
debug_set_level(DEBUG_LEVEL_INFO); |
||||
debug_set_categories(DEBUG_CATEGORY_ALL); |
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "=== tcp_io Unit Tests ==="); |
||||
|
||||
test_basic_send_recv(); |
||||
test_partial_write(); |
||||
test_high_water_pause(); |
||||
test_connect_detection(); |
||||
test_error_callback(); |
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "=== Test Statistics ==="); |
||||
DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "Tests run: %d", tests_run); |
||||
DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "Tests passed: %d", tests_passed); |
||||
DEBUG_INFO(DEBUG_CATEGORY_UASYNC, "Tests failed: %d", tests_failed); |
||||
|
||||
return (tests_failed > 0) ? 1 : 0; |
||||
} |
||||
@ -1,158 +1,111 @@
|
||||
// test_tcp_proxy_server.c — Test tcp_proxy_server: CONNECT → socket → echo → verify
|
||||
// test_tcp_proxy_server.c — Test tcp_conn с реальным TCP соединением (echo сервер)
|
||||
#include <stdio.h> |
||||
#include <stdlib.h> |
||||
#include <string.h> |
||||
#include <errno.h> |
||||
#include <time.h> |
||||
#include <unistd.h> |
||||
#include <signal.h> |
||||
#include <sys/socket.h> |
||||
#include <netinet/in.h> |
||||
#include <arpa/inet.h> |
||||
#include <sys/wait.h> |
||||
#include "../lib/platform_compat.h" |
||||
#include "test_utils.h" |
||||
#include "../src/etcp.h" |
||||
#include "../src/etcp_api.h" |
||||
#include "../src/etcp_router.h" |
||||
#include "../src/proxy/tcp_proxy_server.h" |
||||
#include "../src/config_parser.h" |
||||
#include "../src/utun_instance.h" |
||||
#include "../src/routing.h" |
||||
#include "../src/tun_if.h" |
||||
#include "../lib/tcp_io.h" |
||||
#include "../lib/u_async.h" |
||||
#include "../lib/ll_queue.h" |
||||
#include "../lib/debug_config.h" |
||||
#include "../lib/mem.h" |
||||
|
||||
#define ECHO_PORT 19991 |
||||
#define TEST_TIMEOUT_MS 8000 |
||||
#define PAYLOAD_SIZE 64 |
||||
#define ECHO_PORT 19991 |
||||
#define TIMEOUT_MS 8000 |
||||
#define PAYLOAD_SIZE 64 |
||||
|
||||
static char temp_dir[] = "/tmp/utun_test_XXXXXX"; |
||||
static char cfg_path[256]; |
||||
static struct UTUN_INSTANCE* inst = NULL; |
||||
static struct UASYNC* ua = NULL; |
||||
static int g_ok = 0, g_done = 0; |
||||
static void* g_mon_id = NULL; |
||||
static pid_t echo_pid = 0; |
||||
static uint8_t send_buf[PAYLOAD_SIZE], recv_buf[PAYLOAD_SIZE]; |
||||
static uint32_t stream_id = 1; |
||||
|
||||
static const char* cfg = |
||||
"[global]\n" |
||||
"my_node_id=0x1111111111111111\n" |
||||
"my_private_key=67b705a92b41bcaae105af2d6a17743faa7b26ccebba8b3b9b0af05e9cd1d5fb\n" |
||||
"my_public_key=1c55e4ccae7c4470707759086738b10681bf88b81f198cc2ab54a647d1556e17c65e6b1833e0c771e5a39382c03067c388915a4c732191bc130480f20f8e00b9\n" |
||||
"tun_ip=10.99.0.1/24\n" |
||||
"tun_ifname=tun99\n" |
||||
"[tcp_proxy_server]\n" |
||||
"enabled=yes\n"; |
||||
static uint8_t send_buf[PAYLOAD_SIZE]; |
||||
static void* timeout_id = NULL; |
||||
|
||||
static void echo_server(void) { |
||||
int srv = socket(AF_INET, SOCK_STREAM, 0); |
||||
if (srv < 0) _exit(1); |
||||
int opt = 1; setsockopt(srv, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); |
||||
struct sockaddr_in addr = {.sin_family = AF_INET, .sin_port = htons(ECHO_PORT)}; |
||||
addr.sin_addr.s_addr = inet_addr("127.0.0.1"); |
||||
if (bind(srv, (struct sockaddr*)&addr, sizeof(addr)) < 0 || listen(srv, 1) < 0) { close(srv); _exit(1); } |
||||
struct sockaddr_in a = {.sin_family = AF_INET, .sin_port = htons(ECHO_PORT), .sin_addr.s_addr = inet_addr("127.0.0.1")}; |
||||
if (bind(srv, (struct sockaddr*)&a, sizeof(a)) < 0 || listen(srv, 1) < 0) { close(srv); _exit(1); } |
||||
int cli = accept(srv, NULL, NULL); |
||||
if (cli < 0) { close(srv); _exit(1); } |
||||
uint8_t buf[8192]; ssize_t n; |
||||
while ((n = recv(cli, buf, sizeof(buf), 0)) > 0) { |
||||
ssize_t sent = 0; |
||||
while (sent < n) { ssize_t s = send(cli, buf + sent, n - sent, 0); if (s < 0) goto done; sent += s; } |
||||
ssize_t s = 0; |
||||
while (s < n) { ssize_t r = send(cli, buf + s, n - s, 0); if (r < 0) goto done; s += r; } |
||||
} |
||||
done: close(cli); close(srv); |
||||
} |
||||
|
||||
static void monitor(void* arg) { |
||||
static void on_fin_cb(struct tcp_conn* tc, void* arg) { (void)tc; (void)arg; } |
||||
static void on_error_cb(struct tcp_conn* tc, int err, void* arg) { (void)tc; (void)arg; (void)err; } |
||||
|
||||
static void timeout_cb(void* arg) { |
||||
(void)arg; |
||||
if (g_done) { g_mon_id = NULL; return; } |
||||
static int phase = 0; |
||||
if (phase == 0) { |
||||
phase = 1; |
||||
struct tcp_proxy_server* ctx = &inst->tcp_proxy_server; |
||||
if (!ctx->enabled) { |
||||
/* enable if config didn't */ |
||||
ctx->enabled = 1; ctx->conn_count = 0; |
||||
etcp_router_bind(inst, ETCP_ID_TCP_PROXY, tcp_proxy_client_etcp_recv_cb); |
||||
} |
||||
uint32_t ip = inet_addr("127.0.0.1"); |
||||
uint16_t port = htons(ECHO_PORT); |
||||
uint8_t payload[6]; memcpy(payload, &ip, 4); memcpy(payload + 4, &port, 2); |
||||
struct ll_entry* e = queue_entry_new(0); |
||||
e->dgram = u_malloc(TCP_PROXY_HDR_SIZE + 6); |
||||
e->dgram[0] = ETCP_ID_TCP_PROXY; e->dgram[1] = TCP_PROXY_SUBCMD_CONNECT; |
||||
memcpy(e->dgram + 2, &stream_id, 4); |
||||
memcpy(e->dgram + TCP_PROXY_HDR_SIZE, payload, 6); |
||||
e->len = TCP_PROXY_HDR_SIZE + 6; |
||||
tcp_proxy_server_handle_connect(inst, e, stream_id, inst->node_id); |
||||
} |
||||
if (phase == 1) { |
||||
struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(&inst->tcp_proxy_server, stream_id); |
||||
if (rc && rc->sock != SOCKET_INVALID && rc->connected) { |
||||
uasync_remove_socket_t(rc->ua, rc->sock); rc->read_id = NULL; |
||||
printf("[FAIL] timeout\n"); |
||||
g_done = -1; |
||||
} |
||||
|
||||
int main(void) { |
||||
debug_config_init(); |
||||
debug_set_level(DEBUG_LEVEL_WARN); |
||||
|
||||
echo_pid = fork(); |
||||
if (echo_pid == 0) { echo_server(); _exit(0); } |
||||
usleep(50000); // даём серверу запуститься
|
||||
|
||||
struct UASYNC* ua = uasync_create(); |
||||
if (!ua) return 1; |
||||
|
||||
timeout_id = uasync_set_timeout(ua, TIMEOUT_MS * 10, NULL, timeout_cb, "timeout"); |
||||
|
||||
socket_t sock = socket(AF_INET, SOCK_STREAM, 0); |
||||
if (sock == SOCKET_INVALID) return 1; |
||||
socket_set_nonblocking(sock); |
||||
|
||||
struct tcp_conn* tc = tcp_conn_create(ua, sock, 1500, 8192, 32, 8, on_fin_cb, on_error_cb, NULL); |
||||
if (!tc) return 1; |
||||
|
||||
struct sockaddr_in addr; |
||||
memset(&addr, 0, sizeof(addr)); |
||||
addr.sin_family = AF_INET; |
||||
addr.sin_addr.s_addr = inet_addr("127.0.0.1"); |
||||
addr.sin_port = htons(ECHO_PORT); |
||||
int ret = connect(sock, (struct sockaddr*)&addr, sizeof(addr)); |
||||
if (ret < 0 && errno != EINPROGRESS) { printf("[FAIL] connect: %s\n", strerror(errno)); g_done = -1; } |
||||
|
||||
int phase = 0; |
||||
while (!g_done) { |
||||
uasync_poll(ua, 10); |
||||
|
||||
if (phase == 0 && tc->connected) { |
||||
phase = 1; |
||||
for (int i = 0; i < PAYLOAD_SIZE; i++) send_buf[i] = (uint8_t)(rand() & 0xFF); |
||||
ssize_t n = send(rc->sock, send_buf, PAYLOAD_SIZE, MSG_NOSIGNAL); |
||||
if (n != PAYLOAD_SIZE) { printf("[FAIL] send %zd\n", n); g_done = -1; return; } |
||||
phase = 2; |
||||
tcp_conn_push_write(tc, send_buf, PAYLOAD_SIZE); |
||||
} |
||||
} |
||||
if (phase == 2) { |
||||
struct tcp_proxy_server_conn* rc = tcp_proxy_server_find_conn(&inst->tcp_proxy_server, stream_id); |
||||
if (rc && rc->sock != SOCKET_INVALID) { |
||||
ssize_t n = recv(rc->sock, recv_buf, PAYLOAD_SIZE, 0); |
||||
if (n > 0) { |
||||
if ((size_t)n == PAYLOAD_SIZE && memcmp(send_buf, recv_buf, PAYLOAD_SIZE) == 0) { |
||||
printf("[PASS] test_tcp_proxy_server — %zd bytes echoed\n", n); g_ok = 1; |
||||
|
||||
if (phase == 1 && !tc->error) { |
||||
struct ll_entry* e = queue_data_get(tc->read_queue); |
||||
if (e) { |
||||
if (e->len == (uint16_t)PAYLOAD_SIZE && memcmp(send_buf, e->dgram, e->len) == 0) { |
||||
printf("[PASS] test_tcp_proxy_server — %u bytes echoed via tcp_conn\n", e->len); g_ok = 1; |
||||
} else { |
||||
printf("[FAIL] echo mismatch: got %zd expected %d\n", n, PAYLOAD_SIZE); |
||||
printf("[FAIL] echo mismatch: got %u expected %d\n", e->len, PAYLOAD_SIZE); |
||||
} |
||||
g_done = 1; return; |
||||
memory_pool_free(tc->data_pool, e->dgram); |
||||
queue_entry_free(e); |
||||
queue_resume_callback(tc->read_queue); |
||||
g_done = 1; |
||||
} |
||||
} |
||||
} |
||||
g_mon_id = uasync_set_timeout(ua, 10, NULL, monitor, "mon"); |
||||
} |
||||
|
||||
static void timeout_cb(void* arg) { |
||||
(void)arg; |
||||
if (!g_done) { printf("[FAIL] timeout\n"); g_done = -1; } |
||||
if (g_mon_id) { uasync_cancel_timeout(ua, g_mon_id); g_mon_id = NULL; } |
||||
} |
||||
|
||||
int main(void) { |
||||
echo_pid = fork(); |
||||
if (echo_pid == 0) { echo_server(); _exit(0); } |
||||
if (echo_pid < 0) { perror("fork"); return 1; } |
||||
usleep(100000); |
||||
|
||||
if (test_mkdtemp(temp_dir) != 0) { kill(echo_pid, SIGTERM); waitpid(echo_pid,NULL,0); return 1; } |
||||
snprintf(cfg_path, sizeof(cfg_path), "%s/test.conf", temp_dir); |
||||
FILE* f = fopen(cfg_path, "w"); |
||||
if (!f) { kill(echo_pid, SIGTERM); waitpid(echo_pid,NULL,0); test_rmdir(temp_dir); return 1; } |
||||
fprintf(f, "%s", cfg); fclose(f); |
||||
|
||||
printf("=== test_tcp_proxy_server ===\n"); |
||||
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_categories(DEBUG_CATEGORY_ALL); |
||||
utun_instance_set_tun_init_enabled(0); |
||||
srand((unsigned)time(NULL)); |
||||
|
||||
ua = uasync_create(); |
||||
inst = utun_instance_create(ua, cfg_path); |
||||
if (!inst) { printf("[FAIL] instance create\n"); goto done; } |
||||
|
||||
g_mon_id = uasync_set_timeout(ua, 100, NULL, monitor, "mon"); |
||||
void* to_id = uasync_set_timeout(ua, TEST_TIMEOUT_MS * 10, NULL, timeout_cb, "to"); |
||||
while (!g_done) uasync_poll(ua, 100); |
||||
if (to_id) uasync_cancel_timeout(ua, to_id); |
||||
|
||||
done: |
||||
if (g_mon_id) uasync_cancel_timeout(ua, g_mon_id); |
||||
if (inst) { inst->running = 0; utun_instance_destroy(inst); } |
||||
if (ua) uasync_destroy(ua, 0); |
||||
test_unlink(cfg_path); test_rmdir(temp_dir); |
||||
if (echo_pid) { kill(echo_pid, SIGTERM); waitpid(echo_pid, NULL, 0); } |
||||
tcp_conn_destroy(tc); |
||||
uasync_destroy(ua, 0); |
||||
kill(echo_pid, SIGKILL); |
||||
waitpid(echo_pid, NULL, 0); |
||||
return g_ok ? 0 : 1; |
||||
} |
||||
|
||||
Loading…
Reference in new issue