Keep local callback context private and fully detach services
This commit is contained in:
evgeny committed
1 parent
066b895055
commit
c6f28571b7
2 files changed
+50
-15
No files matched your search
@@ -436,15 +436,6 @@ static void router_confirm_peer(struct ETCP_ROUTER_CONN* rconn, const struct SVC
|
||||
if (!rconn->closed) router_send_kick(rconn);
|
||||
}
|
||||
|
||||
// Синтетический loopback-conn (static) для доставки сервисного коллбэка при локальных событиях
|
||||
// (CLOSE, restart) без реального сетевого соединения.
|
||||
static struct ETCP_CONN* loopback_conn(struct UTUN_INSTANCE* inst) {
|
||||
static struct ETCP_CONN c;
|
||||
memset(&c, 0, sizeof(c));
|
||||
c.instance = inst; c.peer_node_id = inst->node_id;
|
||||
return &c;
|
||||
}
|
||||
|
||||
// Доставить локальному сервису событие закрытия conn (пустая кодограмма ROUTER_SVC_HDR_SIZE).
|
||||
static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn) {
|
||||
struct UTUN_INSTANCE* inst = rconn->inst;
|
||||
@@ -462,7 +453,8 @@ static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn) {
|
||||
e->len = ROUTER_SVC_HDR_SIZE;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_close_svc: svc_id=%u remote=%016llx",
|
||||
rconn->svc_id, (unsigned long long)rconn->remote_node_id);
|
||||
cb(loopback_conn(rconn->inst), e);
|
||||
struct ETCP_CONN local = {.instance = inst, .peer_node_id = inst->node_id};
|
||||
cb(&local, e);
|
||||
}
|
||||
|
||||
// Финальный шаг закрытия: освобождает ll entry (вызывается через uasync_call_soon, уже вне очередей).
|
||||
@@ -1050,7 +1042,8 @@ static void router_deliver_loopback(struct UTUN_INSTANCE* inst, uint64_t group_i
|
||||
if (payload_len > 0) memcpy(e->dgram + ROUTER_SVC_PAYLOAD_OFF, entry->dgram + 1, payload_len);
|
||||
e->len = ROUTER_SVC_HDR_SIZE + payload_len;
|
||||
queue_dgram_free(entry); queue_entry_free(entry);
|
||||
cb(loopback_conn(inst), e);
|
||||
struct ETCP_CONN local = {.instance = inst, .peer_node_id = inst->node_id};
|
||||
cb(&local, e);
|
||||
}
|
||||
|
||||
// Доставка: декодирует (подпись/шифрование) перед вызовом сервисного коллбэка.
|
||||
@@ -1091,6 +1084,8 @@ static int router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn
|
||||
|
||||
// Собрать доставку из recv_q: последовательно доставлять пакеты по возрастанию rx_seq, пока есть следующий.
|
||||
static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn) {
|
||||
struct ETCP_CONN local = {.instance = rconn->inst, .peer_node_id = rconn->inst->node_id};
|
||||
if (!conn) conn = &local;
|
||||
uint32_t next = rconn->rx_seq;
|
||||
while (1) {
|
||||
struct ll_entry* entry = queue_find_data_by_index(rconn->recv_q, &next);
|
||||
@@ -1112,7 +1107,7 @@ static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN
|
||||
rconn->receive_blocked = 0;
|
||||
queue_remove_data(rconn->recv_q, entry);
|
||||
uint64_t peer_id = rconn->peer_reset_id;
|
||||
int ret = router_deliver(rconn, conn ? conn : loopback_conn(rconn->inst), entry->dgram, entry->len);
|
||||
int ret = router_deliver(rconn, conn, entry->dgram, entry->len);
|
||||
if (ret != 0 || rconn->closed || rconn->peer_reset_id != peer_id) { free_entry(entry); return; }
|
||||
rconn->rx_seq = next + 1;
|
||||
next++;
|
||||
@@ -1435,12 +1430,13 @@ void etcp_router_destroy(struct UTUN_INSTANCE* inst) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "etcp_router destroyed");
|
||||
}
|
||||
|
||||
// Зарегистрировать обработчик сервиса. callback=NULL — снять обработчик (эквивалент unbind без ошибки).
|
||||
// Зарегистрировать обработчик; callback=NULL снимает callback, readiness и crypto policy без ошибки отсутствия.
|
||||
int etcp_router_bind(struct UTUN_INSTANCE* inst, uint8_t svc_id, etcp_recv_fn callback) {
|
||||
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL instance"); return -1; }
|
||||
if (!callback) {
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "unbind svc_id=%u", svc_id);
|
||||
inst->router_bindings.callbacks[svc_id] = NULL;
|
||||
inst->router_bindings.receive_ready[svc_id] = NULL;
|
||||
inst->router_bindings.required_crypto[svc_id] = 0;
|
||||
return 0;
|
||||
}
|
||||
@@ -1486,7 +1482,7 @@ static void router_receive_pump(void* arg) {
|
||||
rconn->receive_pump = NULL;
|
||||
if (rconn->closed) return;
|
||||
// Физический recv_conn мог исчезнуть за время дискового I/O; сервис получает стабильный instance.
|
||||
router_try_assembly(rconn, loopback_conn(rconn->inst));
|
||||
router_try_assembly(rconn, NULL);
|
||||
if (!rconn->closed) router_schedule_ack(rconn);
|
||||
}
|
||||
|
||||
|
||||
@@ -1401,8 +1401,10 @@ static int test_required_crypto(void) {
|
||||
TESTASSERT(!c->tx_seq && !c->send_q->count);
|
||||
TESTASSERT(etcp_router_unbind(&inst, TEST_SVC_ID) == 0 && !inst.router_bindings.required_crypto[TEST_SVC_ID]);
|
||||
TESTASSERT(etcp_router_bind(&inst, TEST_SVC_ID, test_handler) == 0);
|
||||
etcp_router_set_receive_ready(&inst, TEST_SVC_ID, receive_ready);
|
||||
TESTASSERT(etcp_router_require_crypto(&inst, TEST_SVC_ID, ROUTER_FLAG_ENCRYPTED) == 0);
|
||||
TESTASSERT(etcp_router_bind(&inst, TEST_SVC_ID, NULL) == 0 && !inst.router_bindings.required_crypto[TEST_SVC_ID]);
|
||||
TESTASSERT(etcp_router_bind(&inst, TEST_SVC_ID, NULL) == 0 && !inst.router_bindings.required_crypto[TEST_SVC_ID]
|
||||
&& !inst.router_bindings.receive_ready[TEST_SVC_ID]);
|
||||
TEARDOWN(); PASS(); return 0;
|
||||
}
|
||||
|
||||
@@ -1469,10 +1471,47 @@ static int test_reopen_from_close_callback(void) {
|
||||
TEARDOWN(); PASS(); return 0;
|
||||
}
|
||||
|
||||
static struct UTUN_INSTANCE* nested_loopback_target;
|
||||
static unsigned nested_loopback_calls;
|
||||
static int nested_loopback_error;
|
||||
|
||||
/* Вложенная доставка другому ядру не должна менять conn текущего callback. */
|
||||
static void nested_loopback_handler(struct ETCP_CONN* conn, struct ll_entry* entry) {
|
||||
struct UTUN_INSTANCE* original = conn->instance;
|
||||
uint64_t peer = conn->peer_node_id;
|
||||
nested_loopback_calls++;
|
||||
queue_dgram_free(entry); queue_entry_free(entry);
|
||||
if (nested_loopback_target) {
|
||||
struct UTUN_INSTANCE* target = nested_loopback_target; nested_loopback_target = NULL;
|
||||
struct ll_entry* next = ll_alloc_lldgram(2);
|
||||
if (!next) { nested_loopback_error = 1; return; }
|
||||
next->dgram[0] = TEST_SVC_ID; next->dgram[1] = 0x5a; next->len = 2;
|
||||
if (etcp_route_send(target, TOPO_GROUP_UTUN, target->node_id, next, 0, 0)) nested_loopback_error = 1;
|
||||
if (conn->instance != original || conn->peer_node_id != peer) nested_loopback_error = 1;
|
||||
}
|
||||
}
|
||||
|
||||
static int test_nested_loopback(void) {
|
||||
TEST("nested self-delivery preserves callback instance and synthetic peer");
|
||||
struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb;
|
||||
SETUP(); struct UTUN_INSTANCE other = {.ua = ua, .node_id = TEST_REMOTE_NODE};
|
||||
TESTASSERT(!etcp_router_init(&other));
|
||||
TESTASSERT(!etcp_router_bind(&inst, TEST_SVC_ID, nested_loopback_handler));
|
||||
TESTASSERT(!etcp_router_bind(&other, TEST_SVC_ID, nested_loopback_handler));
|
||||
nested_loopback_target = &other; nested_loopback_calls = 0; nested_loopback_error = 0;
|
||||
struct ll_entry* e = ll_alloc_lldgram(2); TESTASSERT(e);
|
||||
e->dgram[0] = TEST_SVC_ID; e->dgram[1] = 0x5a; e->len = 2;
|
||||
int result = etcp_route_send(&inst, TOPO_GROUP_UTUN, inst.node_id, e, 0, 0);
|
||||
etcp_router_destroy(&other); TEARDOWN();
|
||||
TESTASSERT(!result && nested_loopback_calls == 2 && !nested_loopback_error);
|
||||
PASS(); return 0;
|
||||
}
|
||||
|
||||
int main(void) {
|
||||
// Run all tests
|
||||
test_close_group_from_close_callback();
|
||||
test_reopen_from_close_callback();
|
||||
test_nested_loopback();
|
||||
test_init_destroy();
|
||||
test_bind_unbind();
|
||||
test_conn_get_create();
|
||||
|
||||
Reference in new issue
Block a user