Reuse ETCP registry entries and reject duplicate peer indexing
This commit is contained in:
1 parent
fb995db489
commit
c7d7789a58
2 files changed
+43
-31
No files matched your search
+37
-30
@@ -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);
|
||||
|
||||
@@ -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).
|
||||
|
||||
Reference in new issue
Block a user