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

Retain selected relay issuance before delayed READY and ACCEPT

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

15
src/stream/stream_network.c
Unescape View File

@ -1241,13 +1241,14 @@ static struct stream_relay_request *relay_request_new(struct stream_network *net
return r;
}
/* Наличие транспортных handles/INFO/ENDED не является действующей раздачей. */
static unsigned relay_active(struct stream_network *net, const struct stream_descriptor *key, uint64_t peer, int outgoing) {
/* Downstream удерживает приём только в ACTIVE; выбранный upstream удерживает запрос также до ACCEPT, но INFO-пробы — нет. */
static unsigned relay_flows(struct stream_network *net, const struct stream_descriptor *key, uint64_t peer, int outgoing, int pending) {
unsigned count = 0;
for (struct stream_peer *p = net->peers; p; p = p->next) {
if (peer && p->node_id != peer) continue;
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
if (f->outgoing == outgoing && !f->info.probe_only && f->info.state == STREAM_RECEIVE_ACTIVE
if (f->outgoing == outgoing && !f->info.probe_only
&& (f->info.state == STREAM_RECEIVE_ACTIVE || (pending && f->info.state < STREAM_RECEIVE_ACTIVE))
&& same_stream(&f->descriptor, key)) count++;
}
return count;
@ -1307,11 +1308,11 @@ void stream_relay_demand(struct UTUN_INSTANCE *inst, const struct stream_descrip
if (!net || net->stopping || stream_local_get(inst, key, &local) != STREAM_OK) return;
for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) {
if (!r->outgoing || !same_stream(&r->descriptor, key) || r->status >= STREAM_RELAY_REJECTED) continue;
unsigned active = relay_active(net, key, r->peer, 1);
unsigned in_use = relay_flows(net, key, r->peer, 1, 1);
int drained = local.finished && (local.last == UINT64_MAX || (local.anchored && local.next > local.last));
if (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))
else if (local.relay_only || drained || (!needed && !in_use))
relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now);
else return;
}
@ -1442,7 +1443,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) {
&& stream_member_supernode(net->inst, r->descriptor.group_id, r->outgoing ? r->peer : net->inst->node_id) == 1;
if (!valid) relay_finish(net, r, r->outgoing ? STREAM_RELAY_CANCELLED : STREAM_RELAY_REJECTED, STREAM_REASON_MEMBER, now);
}
unsigned active = relay_active(net, &r->descriptor, r->peer, r->outgoing);
unsigned active = relay_flows(net, &r->descriptor, r->peer, r->outgoing, 0);
if (r->status < STREAM_RELAY_REJECTED && active) {
if (r->status != STREAM_RELAY_BOUND) {
r->status = STREAM_RELAY_BOUND; r->dirty = r->urgent = 1; r->progress_ms = now;
@ -1505,7 +1506,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) {
unsigned pending = 0;
for (struct stream_relay_request *r = net->relay_requests; r; r = r->next)
if (!r->outgoing && same_stream(&r->descriptor, &relay->descriptor) && r->status <= STREAM_RELAY_READY) pending++;
unsigned active = relay_active(net, &relay->descriptor, 0, 0);
unsigned active = relay_flows(net, &relay->descriptor, 0, 0, 0);
if (now >= relay->checked_ms) {
relay->checked_ms = now + 1000;
relay->allowed = stream_member_supernode(net->inst, relay->descriptor.group_id, net->inst->node_id) == 1;

40
tests/test_stream_service.c
Unescape View File

@ -55,14 +55,14 @@ static int automatic_drop_tail;
static etcp_recv_fn automatic_handlers[4];
static uint8_t automatic_request[STREAM_WIRE_HEADER + 1];
static etcp_recv_fn relay_raw_handlers[4], relay_route_handler, relay_status_handler;
static unsigned relay_data[3], relay_ends[3], relay_requests_seen, relay_statuses_seen;
static unsigned relay_data[3], relay_ends[3], relay_requests_seen, relay_statuses_seen, relay_cancels_seen;
static int relay_drop_source;
static uint8_t relay_request_wire[STREAM_WIRE_HEADER];
static struct {
int hold_info, hold_request, check_starts;
int hold_info, hold_request, hold_ready, check_starts;
uint8_t info[STREAM_WIRE_HEADER + STREAM_INFO_SIZE + STREAM_PATH_MAX * 8];
size_t info_length;
unsigned source_subscribes, helper_subscribes, accepted;
unsigned source_subscribes, helper_subscribes, accepted, ready_held;
uint64_t request_tb, start;
} relay_race;
static struct {
@ -1045,6 +1045,8 @@ static void relay_test_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
/* Сохраняем реальный подписанный routed-запрос для проверки позднего повтора после освобождения буфера. */
static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
&& entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_CANCEL) relay_cancels_seen++;
if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
&& entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_REQUEST) {
memcpy(relay_request_wire, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, STREAM_WIRE_HEADER); relay_requests_seen++;
@ -1057,7 +1059,15 @@ static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
/* Счётчик STATUS выявляет ответы на дубликаты, не меняя подписанное управление и его порядок. */
static void relay_test_status(struct ETCP_CONN *conn, struct ll_entry *entry) {
if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
&& entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_STATUS) relay_statuses_seen++;
&& entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_STATUS) {
relay_statuses_seen++;
size_t payload = ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER;
if (relay_race.hold_ready && entry->len > payload && entry->dgram[payload] == STREAM_RELAY_READY) {
relay_race.ready_held++;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "bootstrap test holds READY ordinal=%u", relay_race.ready_held);
queue_dgram_free(entry); queue_entry_free(entry); return;
}
}
relay_status_handler(conn, entry);
}
@ -1150,19 +1160,33 @@ static void test_supernode_relay(void) {
CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
test_clock_run_for(ua, 500); stream_close(viewers[2]); viewers[2] = NULL;
test_clock_run_for(ua, 2000); CHECK(relay_requests_seen == requests_before);
memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_info = relay_race.check_starts = 1;
/* Merkle объявляет буфер помощника раньше STATUS READY: выбранная выдача должна удержать STARTING-запрос. */
memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_ready = 1;
relay_drop_source = 1;
unsigned cancellations_before = relay_cancels_seen;
unsigned expected_data = relay_data[2] + 1;
CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
deadline = get_time_tb() + 40000;
while (relay_data[2] < expected_data && relay_cancels_seen == cancellations_before && get_time_tb() < deadline)
test_clock_poll(ua, 100);
CHECK(relay_race.ready_held && relay_cancels_seen == cancellations_before);
CHECK(relay_data[2] == expected_data && relay_race.helper_subscribes);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] helper INFO before delayed READY keeps selected pending issuance; no premature CANCEL");
stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 40000);
CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT);
memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_info = relay_race.check_starts = 1;
started = get_time_tb();
expected_data = relay_data[2] + 1;
CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK);
deadline = get_time_tb() + 300000;
while (relay_data[2] < 3 && get_time_tb() < deadline) test_clock_poll(ua, 100);
CHECK(relay_data[2] == 3 && relay_requests_seen && !relay_data[1]);
while (relay_data[2] < expected_data && get_time_tb() < deadline) test_clock_poll(ua, 100);
CHECK(relay_data[2] == expected_data && relay_requests_seen && !relay_data[1]);
CHECK(get_time_tb() - started < 10000 && relay_race.request_tb - started >= 990);
uint64_t connected_ms = (get_time_tb() - started) / 10;
test_clock_run_for(ua, 1000); CHECK(relay_race.accepted == 255);
CHECK(relay_race.helper_subscribes && !relay_race.source_subscribes);
relay_release_info(); test_clock_run_for(ua, 1000);
CHECK(!relay_race.source_subscribes && relay_data[2] == 3);
CHECK(!relay_race.source_subscribes && relay_data[2] == expected_data);
struct stream_message duplicate; CHECK(stream_wire_parse(relay_request_wire, sizeof(relay_request_wire), &duplicate) == STREAM_OK);
CHECK(duplicate.receiver_epoch);
unsigned statuses_before = relay_statuses_seen;

Write Preview
Loading…
Cancel
Save