diff --git a/src/transport_layer/stcp_link.c b/src/transport_layer/stcp_link.c index 4c57dbee..95c5fec9 100644 --- a/src/transport_layer/stcp_link.c +++ b/src/transport_layer/stcp_link.c @@ -42,10 +42,8 @@ struct stcp_link { void (*on_close_cb)(struct stcp_link *link, int err, void *arg); void *close_arg; - struct ll_queue *tx_queue; // owned by this link - - struct ll_queue *saved_rx_queue; // saved rx_queue for pre-closed conn cleanup - uint8_t conn_pre_closed; // 1 = conn already CLOSED before stcp_link_close + struct ll_queue *rx_queue; // Обе очереди принадлежат адаптеру, conn только заимствует их. + struct ll_queue *tx_queue; uint64_t peer_reset_id; // received from peer during handshake (reset epoch) uint8_t peer_device_type; // CLIENT_TYPE_* from peer handshake @@ -56,6 +54,7 @@ struct stcp_link { static void link_rx_cb(struct ll_queue *q, void *arg) { struct stcp_link *link = (struct stcp_link *)arg; + if (link->closing) return; struct ll_entry *e = queue_data_get(q); if (!e) { queue_resume_callback(q); return; } @@ -88,19 +87,45 @@ static void link_rx_cb(struct ll_queue *q, void *arg) { inst->api_bindings.callbacks[ETCP_ID_DATA](link->etcp_conn ? link->etcp_conn : NULL, e); else { queue_dgram_free(e); queue_entry_free(e); } } - queue_resume_callback(q); - if (link->conn) stcp_rx_resume_if_needed(link->conn); + if (!link->closing) { + queue_resume_callback(q); + if (link->conn) stcp_rx_resume_if_needed(link->conn); + } } // ====== STCP close → link down ====== static void stcp_link_on_stcp_close(struct stcp_conn *conn, int err, void *arg) { struct stcp_link *link = (struct stcp_link *)arg; - (void)conn; (void)err; callring_put(CR_SLINK_ONSTCP, (uintptr_t)conn, (uintptr_t)arg); link->ready = 0; - if (link->etcp_link) { link->etcp_link->recv_keepalive = 0; link->etcp_link->link_status = 0; } - if (link->on_close_cb) link->on_close_cb(link, err, link->close_arg); + link->conn = NULL; /* Серверный conn освобождается после возврата из этого callback. */ + conn->rx_queue = conn->tx_queue = NULL; + queue_set_callback(link->rx_queue, NULL, NULL); + queue_set_callback(link->tx_queue, NULL, NULL); + void (*callback)(struct stcp_link*, int, void*) = link->on_close_cb; + void *callback_arg = link->close_arg; + link->on_close_cb = NULL; link->close_arg = NULL; + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "stcp adapter closed: link=%p error=%d cancelling=%u", (void*)link, err, link->closing); + if (!link->closing && callback) callback(link, err, callback_arg); + /* Callback мог вызвать close; после уведомления адаптер больше не используем. */ +} + +static int stcp_link_init_queues(struct stcp_link *link) { + link->rx_queue = queue_new(link->ua, 0, 0, 0, "stcp_rx"); + link->tx_queue = queue_new(link->ua, 0, 0, 0, "stcp_tx"); + if (!link->rx_queue || !link->tx_queue) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP adapter queue allocation failed link=%p RX=%p TX=%p", + (void*)link, (void*)link->rx_queue, (void*)link->tx_queue); + return -1; + } + queue_set_callback(link->rx_queue, link_rx_cb, link); + queue_set_threshold(link->rx_queue, STCP_RX_QUEUE_MAX_PACKETS, STCP_RX_QUEUE_MAX_BYTES); + queue_set_threshold(link->tx_queue, 0, 0); + queue_set_waiter_defer(link->tx_queue, 1); + stcp_conn_set_rx_queue(link->conn, link->rx_queue); + stcp_conn_set_tx_queue(link->conn, link->tx_queue); + return 0; } // ====== server accept → link ====== @@ -108,7 +133,7 @@ static void stcp_link_on_stcp_close(struct stcp_conn *conn, int err, void *arg) static void server_accept_cb(struct stcp_conn *conn, void *arg) { struct stcp_server *ss = (struct stcp_server *)arg; struct stcp_link *link = u_calloc(1, sizeof(struct stcp_link)); - if (!link) { stcp_conn_free(conn); return; } + if (!link) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP server adapter allocation failed"); stcp_conn_do_close(conn, ENOMEM); return; } link->inst = ss->inst; link->ua = ss->inst->ua; link->ready = 1; @@ -119,14 +144,7 @@ static void server_accept_cb(struct stcp_conn *conn, void *arg) { stcp_conn_set_on_close(conn, stcp_link_on_stcp_close, link); - struct ll_queue *rx = queue_new(conn->ua, 0, 0, 0, "srx"); - queue_set_callback(rx, link_rx_cb, link); - queue_set_threshold(rx, STCP_RX_QUEUE_MAX_PACKETS, STCP_RX_QUEUE_MAX_BYTES); - stcp_conn_set_rx_queue(conn, rx); - link->tx_queue = queue_new(conn->ua, 0, 0, 0, "stx"); - queue_set_threshold(link->tx_queue, 0, 0); - queue_set_waiter_defer(link->tx_queue, 1); - stcp_conn_set_tx_queue(conn, link->tx_queue); + if (stcp_link_init_queues(link) < 0) { stcp_link_close(link); return; } DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_link: server accepted connection"); if (ss->on_link) ss->on_link(link, ss->tcp_sock); @@ -136,23 +154,14 @@ static void server_accept_cb(struct stcp_conn *conn, void *arg) { static void client_ready_cb(struct stcp_conn *conn, void *arg) { struct stcp_link *link = (struct stcp_link *)arg; - if (!conn) return; - link->ready = 1; + if (!conn || link->closing) return; link->conn = conn; link->peer_reset_id = conn->peer_reset_id; link->peer_device_type = conn->peer_device_type; link->peer_keepalive_interval = conn->peer_keepalive_interval; - stcp_conn_set_on_close(conn, stcp_link_on_stcp_close, link); - - struct ll_queue *rx = queue_new(conn->ua, 0, 0, 0, "crx"); - queue_set_callback(rx, link_rx_cb, link); - queue_set_threshold(rx, STCP_RX_QUEUE_MAX_PACKETS, STCP_RX_QUEUE_MAX_BYTES); - stcp_conn_set_rx_queue(conn, rx); - link->tx_queue = queue_new(conn->ua, 0, 0, 0, "ctx"); - queue_set_threshold(link->tx_queue, 0, 0); - queue_set_waiter_defer(link->tx_queue, 1); - stcp_conn_set_tx_queue(conn, link->tx_queue); + if (stcp_link_init_queues(link) < 0) { stcp_conn_do_close(conn, ENOMEM); return; } + link->ready = 1; { const struct sockaddr_storage *ra = link->etcp_link ? &link->etcp_link->remote_addr : NULL; @@ -163,6 +172,7 @@ static void client_ready_cb(struct stcp_conn *conn, void *arg) { (unsigned long long)link->peer_reset_id); } if (link->etcp_link) etcp_link_enter_ready_tcp(link->etcp_link); + if (link->closing || !link->etcp_conn || link->etcp_conn->close_requested) return; if (link->on_ready_cb) link->on_ready_cb(link, link->ready_arg); } @@ -272,6 +282,13 @@ struct stcp_link *stcp_link_connect(struct ETCP_LINK *etcp_link, 0, client_ready_cb, link, NULL, NULL, NULL, NULL, local_addr, 0, reality, socks, inst); if (!link->cli) { u_free(link); return NULL; } + link->conn = stcp_client_get_conn(link->cli); + if (!link->conn || link->conn->state == STCP_STATE_CLOSED || link->conn->state == STCP_STATE_ERROR) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP adapter startup failed before return target=%s:%u", addr_str, rport); + stcp_client_destroy(link->cli); u_free(link); return NULL; + } + /* Установить до первого poll: закрытие connect/handshake также должно достигать владельца. */ + stcp_conn_set_on_close(link->conn, stcp_link_on_stcp_close, link); DEBUG_INFO(DEBUG_CATEGORY_ETCP, "stcp_link: connecting to %s:%u pubkey=%016llx bind=%s", addr_str, rport, (unsigned long long)*(const uint64_t*)pubkey, @@ -296,32 +313,15 @@ static void stcp_link_close_impl(void *arg) { struct stcp_link *link = (struct stcp_link *)arg; callring_put(CR_SLINK_IMPL, (uintptr_t)link, (uintptr_t)(link->conn)); - if (link->cli) { - struct stcp_conn *c = stcp_client_get_conn(link->cli); - if (!link->conn_pre_closed) { - if (c && c->rx_queue) { stcp_drain_free_queue(c->rx_queue); c->rx_queue = NULL; } - if (link->tx_queue) { stcp_drain_free_queue(link->tx_queue); link->tx_queue = NULL; } - stcp_client_destroy(link->cli); - } else { - if (link->saved_rx_queue) { stcp_drain_free_queue(link->saved_rx_queue); link->saved_rx_queue = NULL; } - if (link->tx_queue) { stcp_drain_free_queue(link->tx_queue); link->tx_queue = NULL; } - u_free(link->cli); - } - } else if (link->conn) { - if (!link->conn_pre_closed) { - struct stcp_conn *conn = link->conn; - if (conn->rx_queue) { stcp_drain_free_queue(conn->rx_queue); conn->rx_queue = NULL; } - if (link->tx_queue) { stcp_drain_free_queue(link->tx_queue); link->tx_queue = NULL; } - stcp_conn_do_close(conn, 0); - } else { - /* conn уже закрыт и освобождается отложенно — не трогаем link->conn, - * работаем с сохранённой rx_queue. */ - if (link->saved_rx_queue) { stcp_drain_free_queue(link->saved_rx_queue); link->saved_rx_queue = NULL; } - if (link->tx_queue) { stcp_drain_free_queue(link->tx_queue); link->tx_queue = NULL; } - } - link->conn = NULL; + struct stcp_conn *conn = link->conn; + if (conn) { + conn->rx_queue = conn->tx_queue = NULL; + conn->on_close = NULL; conn->close_arg = NULL; } - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "stcp_link_close_impl: link=%p cli=%p conn=%p conn_pre_closed=%d", (void*)link, (void*)link->cli, (void*)link->conn, link->conn_pre_closed); + stcp_drain_free_queue(link->rx_queue); stcp_drain_free_queue(link->tx_queue); + if (link->cli) stcp_client_destroy(link->cli); + else if (conn) stcp_conn_do_close(conn, 0); + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "stcp adapter freed: link=%p client=%u", (void*)link, link->cli != NULL); u_free(link); } @@ -329,6 +329,10 @@ void stcp_link_close(struct stcp_link *link) { if (!link) return; if (link->closing) return; link->closing = 1; + link->ready = 0; + link->on_ready_cb = NULL; link->ready_arg = NULL; + queue_set_callback(link->rx_queue, NULL, NULL); + queue_set_callback(link->tx_queue, NULL, NULL); callring_put(CR_SLINK_CLOSE, (uintptr_t)link, (uintptr_t)(link->etcp_link)); /* Обрыв вперёд-ссылки родителя: после отложенного free (close_impl) * etcp_link->tcp_link не должен остаться висячим. */ @@ -342,44 +346,44 @@ void stcp_link_close(struct stcp_link *link) { link->close_arg = NULL; DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "stcp_link_close: link=%p closing=%d", (void*)link, link->closing); - if (link->cli) { - struct stcp_conn *c = stcp_client_get_conn(link->cli); - if (c) { - if (c->hs_timer) { uasync_cancel_timeout(link->ua, c->hs_timer); c->hs_timer = NULL; } - if (c->state == STCP_STATE_CLOSED || c->state == STCP_STATE_ERROR) { - link->saved_rx_queue = c->rx_queue; c->rx_queue = NULL; link->conn_pre_closed = 1; - } - } - } else if (link->conn) { + if (link->conn) { struct stcp_conn *c = link->conn; if (c->hs_timer) { uasync_cancel_timeout(link->ua, c->hs_timer); c->hs_timer = NULL; } - if (c->state == STCP_STATE_CLOSED || c->state == STCP_STATE_ERROR) { - link->saved_rx_queue = c->rx_queue; c->rx_queue = NULL; link->conn_pre_closed = 1; + if (c->socket_id) { + uasync_set_socket_read(link->ua, c->socket_id, 0); + uasync_set_socket_write(link->ua, c->socket_id, 0); } } - uasync_call_soon(link->ua, link, stcp_link_close_impl); + if (!uasync_call_soon(link->ua, link, stcp_link_close_impl)) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "STCP adapter mandatory teardown scheduling failed link=%p", (void*)link); + abort(); + } } int stcp_link_send(struct stcp_link *link, const uint8_t *data, size_t len) { - if (!link || !link->ready) return -1; + if (!link || link->closing || !link->ready || !link->conn) return -1; struct ll_entry *e = queue_entry_new(0); if (!e) return -1; e->dgram = u_malloc(len ? len : 1); if (!e->dgram) { queue_entry_free(e); return -1; } if (len) memcpy(e->dgram, data, len); e->len = (uint16_t)len; - queue_data_put(link->conn->tx_queue, e); + queue_data_put(link->tx_queue, e); return 0; } int stcp_link_is_ready(struct stcp_link *link) { - return link ? link->ready : 0; + return link && !link->closing ? link->ready : 0; } struct ETCP_CONN *stcp_link_get_etcp_conn(struct stcp_link *link) { return link ? link->etcp_conn : NULL; } +struct ETCP_LINK *stcp_link_get_etcp_link(struct stcp_link *link) { + return link && !link->closing ? link->etcp_link : NULL; +} + void stcp_link_set_etcp_conn(struct stcp_link *link, struct ETCP_CONN *conn) { if (!link) return; link->etcp_conn = conn; diff --git a/src/transport_layer/stcp_link.h b/src/transport_layer/stcp_link.h index 5a7030f7..2d6a722c 100644 --- a/src/transport_layer/stcp_link.h +++ b/src/transport_layer/stcp_link.h @@ -1,7 +1,9 @@ /* stcp_link — адаптер STCP-соединения к ETCP_LINK: handshake, очереди DATA, ready/close callbacks. * Серверный путь создаёт входящие линки, connect запускает исходящий для заданного ETCP_LINK. * send копирует data в tx_queue (0 / -1); ready означает готовность STCP, не READY topo-группы. - * close гасит callback внезапного разрыва и откладывает teardown; после close handle не использовать. + * Адаптер владеет RX/TX очередями; stcp_conn их заимствует до закрытия. + * close сразу запрещает send/RX/READY, гасит callback внезапного разрыва и откладывает teardown; + * повторный close внутри активного callback допустим, после возврата в poll handle не использовать. * API и callbacks — uasync-поток. Указатели get_* заимствованы; адреса peer/local используют static-буферы. */ #ifndef STCP_LINK_H #define STCP_LINK_H @@ -52,11 +54,15 @@ struct stcp_server *stcp_server_list_find_by_sock(struct UTUN_INSTANCE *inst, co struct stcp_link *stcp_link_connect(struct ETCP_LINK *etcp_link, struct sockaddr_storage *remote_addr, uint16_t port); +/* NULL — запуск не состоялся, callback не вызывается. Принятый connect уведомляет также об ошибке до READY. + * Все уведомления после возврата connect; владелец устанавливает callbacks до следующего poll. */ void stcp_link_close(struct stcp_link *link); int stcp_link_send(struct stcp_link *link, const uint8_t *data, size_t len); int stcp_link_is_ready(struct stcp_link *link); struct ETCP_CONN *stcp_link_get_etcp_conn(struct stcp_link *link); +/* Заимствованный физический владелец; NULL после close. Проверять после callback до доступа к ETCP_LINK. */ +struct ETCP_LINK *stcp_link_get_etcp_link(struct stcp_link *link); void stcp_link_set_etcp_conn(struct stcp_link *link, struct ETCP_CONN *conn); void stcp_link_set_etcp_link(struct stcp_link *link, struct ETCP_LINK *elink); const struct sockaddr_storage *stcp_link_get_remote_addr(struct stcp_link *link);