jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Apply router backpressure and preserve reusable relay attempts

master
evgeny 6 days ago
parent
e73fab0a73
commit
bbd13215a4
2 changed files with 40 additions and 14 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 48
      src/stream/stream_network.c
  2. 6
      tests/test_stream_service.c

48
src/stream/stream_network.c
Unescape View File

@ -32,7 +32,8 @@ struct stream_relay_request {
struct stream_relay_request *next; struct stream_relay_request *next;
struct stream_descriptor descriptor; struct stream_descriptor descriptor;
uint64_t peer, id, epoch, server_epoch, deadline_ms, retain_ms, sent_ms, checked_ms, progress_ms, progress_seq; uint64_t peer, id, epoch, server_epoch, deadline_ms, retain_ms, sent_ms, checked_ms, progress_ms, progress_seq;
int outgoing, status, reason, dirty, granted; int outgoing, status, reason, dirty, granted, blocked;
struct queue_waiter_handle waiter;
uint8_t end[STREAM_END_SIZE]; uint8_t end[STREAM_END_SIZE];
}; };
struct stream_relay { struct stream_relay {
@ -165,6 +166,12 @@ static void send_ready(struct ll_queue *queue, void *arg) {
(void)queue; struct stream_peer *p = arg; p->wait_q = NULL; peer_schedule(p); (void)queue; struct stream_peer *p = arg; p->wait_q = NULL; peer_schedule(p);
} }
/* Waiter только будит отправку актуального состояния запроса; DATA и старые STATUS в нём не копируются. */
static void relay_send_ready(struct ll_queue *queue, void *arg) {
(void)queue; struct stream_relay_request *r = arg;
r->blocked = 0; r->dirty = 1; r->sent_ms = 0;
}
/* Сохранённый указатель очереди используется только пока waiter действительно зарегистрирован. */ /* Сохранённый указатель очереди используется только пока waiter действительно зарегистрирован. */
static void peer_cancel_wait(struct stream_peer *p) { static void peer_cancel_wait(struct stream_peer *p) {
if (p->waiter.internal || p->waiter.call_soon_id) if (p->waiter.internal || p->waiter.call_soon_id)
@ -198,7 +205,8 @@ static int control_put(struct stream_peer *p, struct stream_message m, const uin
} }
/* Routed-управление обязательно подписано end-to-end, DATA этим путём не передаётся. */ /* Routed-управление обязательно подписано end-to-end, DATA этим путём не передаётся. */
static int route_control(struct stream_network *net, uint64_t peer, struct stream_message m, const uint8_t *payload, size_t length) { static int route_control(struct stream_network *net, uint64_t peer, struct stream_message m, const uint8_t *payload, size_t length,
struct stream_relay_request *request) {
struct UTUN_INSTANCE *inst = net->inst; struct UTUN_INSTANCE *inst = net->inst;
int route = etcp_router_has_route(inst, m.descriptor.group_id, peer); int route = etcp_router_has_route(inst, m.descriptor.group_id, peer);
if (!route) { if (!route) {
@ -211,7 +219,13 @@ static int route_control(struct stream_network *net, uint64_t peer, struct strea
} }
if (!etcp_router_send_q_has_room(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM)) { if (!etcp_router_send_q_has_room(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM)) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for queue room peer=%llx command=%u", DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for queue room peer=%llx command=%u",
(unsigned long long)peer, m.command); return STREAM_WAIT; (unsigned long long)peer, m.command);
if (request) {
request->blocked = 1;
etcp_router_on_send_ready(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM, &request->waiter,
relay_send_ready, request);
}
return STREAM_WAIT;
} }
struct ll_entry *e = ll_alloc_lldgram((uint16_t)(1 + STREAM_WIRE_HEADER + length)); struct ll_entry *e = ll_alloc_lldgram((uint16_t)(1 + STREAM_WIRE_HEADER + length));
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream routed control allocation failed"); return STREAM_ERR_MEMORY; } if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream routed control allocation failed"); return STREAM_ERR_MEMORY; }
@ -251,7 +265,7 @@ static void flow_close(struct STREAM_RECEIVE *f, int reason, int notify) {
.command = f->outgoing ? STREAM_CMD_UNSUBSCRIBE : STREAM_CMD_CLOSED, .command = f->outgoing ? STREAM_CMD_UNSUBSCRIBE : STREAM_CMD_CLOSED,
.sender_epoch = p->own_epoch, .receiver_epoch = p->other_epoch}; .sender_epoch = p->own_epoch, .receiver_epoch = p->other_epoch};
if (p->up) control_put(p, m, &payload, 1); if (p->up) control_put(p, m, &payload, 1);
else if (!f->outgoing) route_control(p->network, p->node_id, m, &payload, 1); else if (!f->outgoing) route_control(p->network, p->node_id, m, &payload, 1, NULL);
} }
f->info.state = STREAM_RECEIVE_CLOSED; f->info.reason = reason; f->info.state = STREAM_RECEIVE_CLOSED; f->info.reason = reason;
if (reason != STREAM_REASON_STOPPED) if (reason != STREAM_REASON_STOPPED)
@ -329,7 +343,7 @@ static void ncd_changed(struct NODE_CONN_DIRECT *handle, enum ncd_event event, v
struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle,
.subblock = f->subblock, .command = STREAM_CMD_CLOSED, .subblock = f->subblock, .command = STREAM_CMD_CLOSED,
.sender_epoch = p->own_epoch, .receiver_epoch = p->other_epoch}; .sender_epoch = p->own_epoch, .receiver_epoch = p->other_epoch};
route_control(p->network, p->node_id, m, &reason, 1); route_control(p->network, p->node_id, m, &reason, 1, NULL);
} }
peer_reset(p, STREAM_REASON_TRANSPORT, STREAM_RESET_ROTATE, NULL); peer_reset(p, STREAM_REASON_TRANSPORT, STREAM_RESET_ROTATE, NULL);
} else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream NCD initial timeout peer=%llx", (unsigned long long)p->node_id); } else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream NCD initial timeout peer=%llx", (unsigned long long)p->node_id);
@ -985,7 +999,7 @@ static void network_tick(void *arg) {
if (!p->up && pending && (!p->reverse_at || now - p->reverse_at >= 10000)) { if (!p->up && pending && (!p->reverse_at || now - p->reverse_at >= 10000)) {
struct stream_message m = {.descriptor = pending->descriptor, .command = STREAM_CMD_REVERSE_REQ, struct stream_message m = {.descriptor = pending->descriptor, .command = STREAM_CMD_REVERSE_REQ,
.sender_epoch = p->own_epoch}; .sender_epoch = p->own_epoch};
if (route_control(net, p->node_id, m, NULL, 0) == STREAM_OK) p->reverse_at = now; if (route_control(net, p->node_id, m, NULL, 0, NULL) == STREAM_OK) p->reverse_at = now;
} }
p->busy--; peer_sweep(p); p->busy--; peer_sweep(p);
if (p->idle && now - p->idle >= STREAM_LEASE_TB && !p->control->count) peer_free(p); if (p->idle && now - p->idle >= STREAM_LEASE_TB && !p->control->count) peer_free(p);
@ -1036,7 +1050,8 @@ void stream_network_destroy(struct stream_network *net) {
stream_close(relay->owner); u_free(relay); stream_close(relay->owner); u_free(relay);
} }
while (net->relay_requests) { while (net->relay_requests) {
struct stream_relay_request *r = net->relay_requests; net->relay_requests = r->next; u_free(r); struct stream_relay_request *r = net->relay_requests; net->relay_requests = r->next;
etcp_router_cancel_send_ready(net->inst, r->descriptor.group_id, r->peer, ETCP_RT_ID_STREAM, &r->waiter); u_free(r);
} }
etcp_remove_conn_status_cbk(net->inst, transport_changed, net); etcp_remove_conn_status_cbk(net->inst, transport_changed, net);
etcp_unbind(net->inst, ETCP_ID_STREAM); etcp_router_unbind(net->inst, ETCP_RT_ID_STREAM); etcp_unbind(net->inst, ETCP_ID_STREAM); etcp_router_unbind(net->inst, ETCP_RT_ID_STREAM);
@ -1214,6 +1229,8 @@ static unsigned relay_active(struct stream_network *net, const struct stream_des
/* Неуспешная помощь закрывает только выдачи своего пира и трансляции, не общий ETCP. */ /* Неуспешная помощь закрывает только выдачи своего пира и трансляции, не общий ETCP. */
static void relay_finish(struct stream_network *net, struct stream_relay_request *r, int status, int reason, uint64_t now) { static void relay_finish(struct stream_network *net, struct stream_relay_request *r, int status, int reason, uint64_t now) {
if (r->status >= STREAM_RELAY_REJECTED) return; if (r->status >= STREAM_RELAY_REJECTED) return;
etcp_router_cancel_send_ready(net->inst, r->descriptor.group_id, r->peer, ETCP_RT_ID_STREAM, &r->waiter);
memset(&r->waiter, 0, sizeof(r->waiter)); r->blocked = 0;
r->status = status; r->reason = reason; r->retain_ms = now + STREAM_RELAY_RETAIN_MS; r->dirty = 1; r->status = status; r->reason = reason; r->retain_ms = now + STREAM_RELAY_RETAIN_MS; r->dirty = 1;
if (r->outgoing) { if (r->outgoing) {
struct stream_peer *p = peer_find(net, r->peer); struct stream_peer *p = peer_find(net, r->peer);
@ -1265,8 +1282,9 @@ void stream_relay_demand(struct UTUN_INSTANCE *inst, const struct stream_descrip
if (!r->outgoing || !same_stream(&r->descriptor, key) || r->status >= STREAM_RELAY_REJECTED) continue; if (!r->outgoing || !same_stream(&r->descriptor, key) || r->status >= STREAM_RELAY_REJECTED) continue;
unsigned active = relay_active(net, key, r->peer, 1); unsigned active = relay_active(net, key, r->peer, 1);
int drained = local.finished && (local.last == UINT64_MAX || (local.anchored && local.next > local.last)); int drained = local.finished && (local.last == UINT64_MAX || (local.anchored && local.next > local.last));
if (local.relay_only || drained || (!needed && !active) if (needed && r->status == STREAM_RELAY_BOUND && now - r->progress_ms >= STREAM_RELAY_START_MS)
|| (needed && r->status == STREAM_RELAY_BOUND && now - r->progress_ms >= STREAM_RELAY_START_MS)) relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_TRANSPORT, now);
else if (local.relay_only || drained || (!needed && !active))
relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now); relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now);
else return; else return;
} }
@ -1280,7 +1298,8 @@ void stream_relay_demand(struct UTUN_INSTANCE *inst, const struct stream_descrip
if (!peers[i].supernode || peer == inst->node_id || peer == key->source_node_id if (!peers[i].supernode || peer == inst->node_id || peer == key->source_node_id
|| !stream_member_online(inst, key->group_id, peer) || !node_conn_direct_compatible(inst, peer)) continue; || !stream_member_online(inst, key->group_id, peer) || !node_conn_direct_compatible(inst, peer)) continue;
struct stream_relay_request *previous = relay_request_find(net, key, peer, 0, 1); struct stream_relay_request *previous = relay_request_find(net, key, peer, 0, 1);
if (previous) continue; if (previous && (previous->status < STREAM_RELAY_REJECTED || previous->status == STREAM_RELAY_REJECTED
|| previous->reason != STREAM_REASON_STOPPED)) continue;
struct stream_member_state member; struct stream_member_state member;
if (stream_member_load(inst, key->group_id, peer, &member, NULL) != STREAM_OK) continue; if (stream_member_load(inst, key->group_id, peer, &member, NULL) != STREAM_OK) continue;
chosen = peer; break; chosen = peer; break;
@ -1338,7 +1357,7 @@ static void relay_dispatch(struct stream_network *net, uint64_t sender, const st
relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_PROTOCOL, now); relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_PROTOCOL, now);
} else relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now); } else relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now);
} else if (status >= STREAM_RELAY_REJECTED) { } else if (status >= STREAM_RELAY_REJECTED) {
relay_finish(net, r, STREAM_RELAY_CANCELLED, m->payload[1], now); relay_finish(net, r, STREAM_RELAY_REJECTED, m->payload[1], now);
} }
return; return;
} }
@ -1428,7 +1447,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) {
} }
} }
} }
if (!r->dirty && !(r->outgoing && r->status <= STREAM_RELAY_READY && now - r->sent_ms >= 1000)) continue; if (r->blocked || (!r->dirty && !(r->outgoing && r->status <= STREAM_RELAY_READY && now - r->sent_ms >= 1000))) continue;
if (r->sent_ms && now - r->sent_ms < 1000) continue; if (r->sent_ms && now - r->sent_ms < 1000) continue;
struct stream_message m = {.descriptor = r->descriptor, .handle = r->id, struct stream_message m = {.descriptor = r->descriptor, .handle = r->id,
.sender_epoch = r->outgoing ? r->epoch : net->relay_epoch, .sender_epoch = r->outgoing ? r->epoch : net->relay_epoch,
@ -1442,7 +1461,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) {
if (r->status == STREAM_RELAY_ENDED) { memcpy(payload + 6, r->end, STREAM_END_SIZE); length += STREAM_END_SIZE; } if (r->status == STREAM_RELAY_ENDED) { memcpy(payload + 6, r->end, STREAM_END_SIZE); length += STREAM_END_SIZE; }
} }
r->sent_ms = now; r->sent_ms = now;
if (route_control(net, r->peer, m, payload, length) == STREAM_OK) r->dirty = 0; if (route_control(net, r->peer, m, payload, length, r) == STREAM_OK) r->dirty = 0;
} }
struct stream_relay **link = &net->relays; struct stream_relay **link = &net->relays;
while (*link) { while (*link) {
@ -1469,7 +1488,8 @@ static void relay_tick(struct stream_network *net, uint64_t now) {
while (*request_link) { while (*request_link) {
struct stream_relay_request *r = *request_link; struct stream_relay_request *r = *request_link;
if (r->status < STREAM_RELAY_REJECTED || now < r->retain_ms) { request_link = &r->next; continue; } if (r->status < STREAM_RELAY_REJECTED || now < r->retain_ms) { request_link = &r->next; continue; }
*request_link = r->next; net->relay_request_count--; u_free(r); *request_link = r->next; net->relay_request_count--;
etcp_router_cancel_send_ready(net->inst, r->descriptor.group_id, r->peer, ETCP_RT_ID_STREAM, &r->waiter); u_free(r);
} }
} }

6
tests/test_stream_service.c
Unescape View File

@ -1087,6 +1087,12 @@ static void test_supernode_relay(void) {
test_clock_run_for(ua, 80000); test_clock_run_for(ua, 80000);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK); CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK);
/* Локальный viewer меняет :r на обычную подписку и делит существующий буфер. */ /* Локальный viewer меняет :r на обычную подписку и делит существующий буфер. */
CHECK(sqlite3_exec(nodes[1]->topo_sqlite_db, "CREATE TRIGGER reject_relay_viewer BEFORE UPDATE OF userinfo ON peers_710002 "
"BEGIN SELECT RAISE(ABORT,'injected relay viewer failure'); END", NULL, NULL, NULL) == SQLITE_OK);
struct STREAM_HANDLE *failed = NULL;
CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &failed) < 0 && !failed);
CHECK(sqlite3_exec(nodes[1]->topo_sqlite_db, "DROP TRIGGER reject_relay_viewer", NULL, NULL, NULL) == SQLITE_OK);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && local.relay_only);
CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &viewers[1]) == STREAM_OK); CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &viewers[1]) == STREAM_OK);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && !local.relay_only); CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && !local.relay_only);
unsigned before = relay_data[1]; unsigned before = relay_data[1];

Write Preview
Loading…
Cancel
Save