diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index c06ab05e..f016e429 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -639,39 +639,46 @@ void etcp_conn_ready(struct ETCP_CONN* conn) { etcp_conn_wake_sender(conn); } -// Move from pending to indexed connections, fire ready callbacks +// Все создатели транспорта публикуют peer до callbacks, сохраняя исходную запись реестра. +int etcp_conn_index_peer(struct ETCP_CONN* conn) { + if (!conn || conn->close_requested || !conn->peer_node_id || !conn->conn_queue_entry || + conn->conn_queue != conn->instance->connections) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "cannot index peer: invalid/closing connection conn=%p", (void*)conn); + return -1; + } + struct ll_entry *qe = conn->conn_queue_entry; + struct conn_queue_entry *ce = (struct conn_queue_entry*)qe->data; + struct ll_entry *existing = queue_find_data_by_index(conn->conn_queue, (const uint8_t*)&conn->peer_node_id); + if (existing && existing != qe) { + struct conn_queue_entry *other = (struct conn_queue_entry*)existing->data; + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] duplicate peer=%016llx conn=%p already indexed by conn=%p", + conn->log_name, (unsigned long long)conn->peer_node_id, (void*)conn, (void*)other->conn); + return -1; + } + if (ce->peer_node_id == conn->peer_node_id) return 0; + uint64_t old_id = ce->peer_node_id; + if (queue_remove_data(conn->conn_queue, qe) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot remove registry entry=%p", conn->log_name, (void*)qe); + return -1; + } + conn->conn_queue_entry = NULL; /* put забирает запись и при ошибке освобождает её. */ + ce->peer_node_id = conn->peer_node_id; + if (queue_data_put_with_index(conn->conn_queue, qe) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot index registry entry for peer=%016llx", + conn->log_name, (unsigned long long)conn->peer_node_id); + return -1; + } + conn->conn_queue_entry = qe; + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] registry indexed: conn=%p entry=%p peer=%016llx->%016llx state=%u", + conn->log_name, (void*)conn, (void*)qe, (unsigned long long)old_id, (unsigned long long)conn->peer_node_id, conn->state); + return 0; +} + +// Готовность потока публикуется отдельно от наличия известного пира в реестре. void etcp_conn_queue_set_ready(struct ETCP_CONN* conn) { if (!conn || conn->state != 0) return; - - // Reindex: remove old entry (key=0), add new entry with real peer_node_id - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] queue_set_ready: removing old entry=%p key=0 -> new key=0x%llx", - conn->log_name, (void*)conn->conn_queue_entry, (unsigned long long)conn->peer_node_id); - if (conn->conn_queue && conn->conn_queue_entry) { - queue_remove_data(conn->conn_queue, conn->conn_queue_entry); - queue_entry_free(conn->conn_queue_entry); - } - - struct ll_entry* existing = queue_find_data_by_index(conn->instance->connections, (const uint8_t*)&conn->peer_node_id); - if (existing) { - struct conn_queue_entry* ece = (struct conn_queue_entry*)existing->data; - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, - "[%s] queue_set_ready: DUPLICATE node_id key=0x%016llx already indexed by [%s] conn=%p state=%d (this=%p)", - conn->log_name, (unsigned long long)conn->peer_node_id, - ece->conn ? ece->conn->log_name : "?", (void*)(ece->conn), - ece->conn ? ece->conn->state : -1, (void*)conn); - } - - struct ll_entry* qe = queue_entry_new(sizeof(struct conn_queue_entry)); - if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] failed to alloc queue entry for ready", conn->log_name); return; } - struct conn_queue_entry* ce = (struct conn_queue_entry*)qe->data; - ce->peer_node_id = conn->peer_node_id; - ce->conn = conn; - conn->conn_queue_entry = qe; - conn->conn_queue = conn->instance->connections; + if (etcp_conn_index_peer(conn) < 0) { etcp_connection_close(conn); return; } conn->state = 1; - queue_data_put_with_index(conn->instance->connections, qe); - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] queue_set_ready: new entry=%p put in queue, count=%d", - conn->log_name, (void*)qe, queue_entry_count(conn->instance->connections)); DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] moved to ready queue, state=%d peer_node_id=0x%llx", conn->log_name, conn->state, (unsigned long long)conn->peer_node_id); diff --git a/src/transport_layer/etcp.h b/src/transport_layer/etcp.h index 350982c5..65cec78e 100644 --- a/src/transport_layer/etcp.h +++ b/src/transport_layer/etcp.h @@ -274,7 +274,12 @@ int etcp_conn_schedule_down_close(struct ETCP_CONN* conn); * После state==2 допустим только DELETE; ref_count задерживает освобождение памяти. */ void etcp_connection_close(struct ETCP_CONN* etcp); -void etcp_conn_queue_set_ready(struct ETCP_CONN* conn); // переиндексировать запись по известному peer_node_id +/* Индексировать уже созданный conn по известному peer_node_id до уведомления о линках. + * Переиспользует запись реестра без аллокаций и callbacks; 0 — успех, -1 — ошибка/дубликат. + * При ошибке вызывающий закрывает conn. Не переводит транспорт в READY. */ +int etcp_conn_index_peer(struct ETCP_CONN* conn); +/* Перевести pending в ready и уведомить INIT/UP; при ошибке индекса закрыть conn. */ +void etcp_conn_queue_set_ready(struct ETCP_CONN* conn); /** * @brief Take a reference on the connection (blocks deferred resource free).