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

Send relay state transitions without retry delays

master
evgeny 1 week ago
parent
5def12351f
commit
e708117259
3 changed files with 32 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
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 14
      src/stream/stream.c
  2. 30
      src/stream/stream_network.c
  3. 2
      src/stream/stream_priv.h

14
src/stream/stream.c
Unescape View File

@ -33,7 +33,7 @@ struct stream_session {
uint8_t end[STREAM_END_SIZE];
int end_valid;
unsigned owners, relay_owners, holds;
int anchored, closing, invalidated, finished, end_delivered, reason;
int anchored, decoded, closing, invalidated, finished, end_delivered, reason;
int restarting, restart_gap, notify_close;
};
@ -296,6 +296,14 @@ static void session_deliver(void *arg) {
int result = s->next_seq == UINT64_MAX ? STREAM_WAIT
: stream_buffer_decode(s->buffer, s->next_seq, s->output, stream_input_size(&s->descriptor), &length);
if (result == STREAM_OK) {
if (!s->decoded) {
s->decoded = 1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream first batch reconstructed group=%llx source=%llx epoch=%llx seq=%llu k=%u",
(unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id,
(unsigned long long)s->descriptor.source_epoch, (unsigned long long)s->next_seq,
(unsigned)(s->descriptor.n - s->descriptor.r));
stream_network_wake(service->network);
}
event.type = STREAM_EVENT_DATA; event.first_seq = event.last_seq = s->next_seq++;
event.data = s->output; event.length = length; emit = 1; again = 1;
} else if (result == STREAM_EXPIRED) {
@ -802,7 +810,7 @@ int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor
struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
if (!s || s->descriptor.source_epoch != key->source_epoch) return STREAM_WAIT;
*local = (struct stream_local){.descriptor = s->descriptor, .buffer = s->buffer, .start = s->start_seq, .last = s->last_seq, .next = s->next_seq,
.anchored = s->anchored, .publisher = s->source != NULL, .finished = s->end_valid,
.anchored = s->anchored, .decoded = s->decoded, .publisher = s->source != NULL, .finished = s->end_valid,
.relay_only = !s->source && s->owners == s->relay_owners, .end = s->end_valid ? s->end : NULL};
return STREAM_OK;
}
@ -863,7 +871,7 @@ int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descrip
s->restarting = 1;
}
s->restart_gap = 0;
s->anchored = s->end_delivered = 0; s->passed_before = s->next_seq = s->start_seq = 0;
s->anchored = s->decoded = s->end_delivered = 0; s->passed_before = s->next_seq = s->start_seq = 0;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receiver restart from end group=%llx source=%llx epoch=%llx retained=%u floor=%llu known=%d",
(unsigned long long)key->group_id, (unsigned long long)key->source_node_id,
(unsigned long long)key->source_epoch, local.buffer != NULL, (unsigned long long)s->restart_from, s->restarting); return STREAM_OK;

30
src/stream/stream_network.c
Unescape View File

@ -30,9 +30,10 @@
/* Запросы не владеют буфером: один внутренний handle обслуживает всех получателей точной эпохи. */
struct stream_relay_request {
struct stream_relay_request *next;
struct stream_network *network;
struct stream_descriptor descriptor;
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, blocked;
int outgoing, status, reason, dirty, urgent, granted, blocked;
struct queue_waiter_handle waiter;
uint8_t end[STREAM_END_SIZE];
};
@ -185,6 +186,7 @@ static void send_ready(struct ll_queue *queue, void *arg) {
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;
stream_network_wake(r->network);
}
/* Сохранённый указатель очереди используется только пока waiter действительно зарегистрирован. */
@ -1090,7 +1092,7 @@ void stream_network_cancel(struct stream_network *net, const struct stream_descr
if (!net) return;
for (struct stream_relay_request *r = net->relay_requests; r; r = r->next)
if (same_stream(&r->descriptor, d) && r->status < STREAM_RELAY_REJECTED) {
r->status = STREAM_RELAY_CANCELLED; r->reason = STREAM_REASON_STOPPED; r->dirty = 1;
r->status = STREAM_RELAY_CANCELLED; r->reason = STREAM_REASON_STOPPED; r->dirty = r->urgent = 1;
r->retain_ms = get_time_tb() / 10 + STREAM_RELAY_RETAIN_MS;
}
for (struct stream_peer *p = net->peers; p; p = p->next) {
@ -1099,6 +1101,7 @@ void stream_network_cancel(struct stream_network *net, const struct stream_descr
if (same_stream(&f->descriptor, d)) flow_close(f, STREAM_REASON_STOPPED, 1);
p->busy--; peer_sweep(p);
}
stream_network_wake(net);
}
/* Исключение поставщика/получателя или self закрывает выдачи, сохраняя локальное кольцо остальных трансляций. */
@ -1228,7 +1231,9 @@ static struct stream_relay_request *relay_request_new(struct stream_network *net
struct stream_relay_request *r = u_calloc(1, sizeof(*r));
if (!r) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "relay request allocation failed"); return NULL; }
r->descriptor = *key; r->peer = peer; r->id = id; r->epoch = epoch; r->outgoing = outgoing;
r->status = STREAM_RELAY_STARTING; r->dirty = 1; r->deadline_ms = get_time_tb() / 10 + STREAM_RELAY_START_MS;
r->network = net;
r->status = STREAM_RELAY_STARTING; r->dirty = r->urgent = 1;
r->deadline_ms = get_time_tb() / 10 + STREAM_RELAY_START_MS;
r->next = net->relay_requests; net->relay_requests = r; net->relay_request_count++;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay request opened group=%llx source=%llx epoch=%llx peer=%llx id=%llx out=%d deadline=%llu",
(unsigned long long)key->group_id, (unsigned long long)key->source_node_id, (unsigned long long)key->source_epoch,
@ -1253,7 +1258,7 @@ static void relay_finish(struct stream_network *net, struct stream_relay_request
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 = r->urgent = 1;
if (r->outgoing) {
struct stream_peer *p = peer_find(net, r->peer);
if (p) {
@ -1360,7 +1365,11 @@ static void relay_dispatch(struct stream_network *net, uint64_t sender, const st
if (stream_member_supernode(net->inst, m->descriptor.group_id, sender) != 1) {
relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_MEMBER, now); return;
}
r->server_epoch = m->sender_epoch;
if (!r->server_epoch) {
r->server_epoch = m->sender_epoch; r->dirty = r->urgent = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "relay server generation confirmed peer=%llx id=%llx generation=%llx",
(unsigned long long)sender, (unsigned long long)r->id, (unsigned long long)r->server_epoch);
}
unsigned status = m->payload[0]; uint64_t remaining = stream_read_be(m->payload + 2, 4);
if (status == STREAM_RELAY_READY || status == STREAM_RELAY_BOUND) {
if (!remaining && status == STREAM_RELAY_READY) return;
@ -1415,7 +1424,7 @@ static void relay_dispatch(struct stream_network *net, uint64_t sender, const st
if (!m->receiver_epoch) return;
if (!peer_open(net, sender)) { relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_TRANSPORT, now); return; }
int result = relay_open(net, &m->descriptor);
if (result == STREAM_OK) r->granted = 1;
if (result == STREAM_OK && !r->granted) { r->granted = r->urgent = 1; }
if (result != STREAM_OK && result != STREAM_WAIT)
relay_finish(net, r, STREAM_RELAY_REJECTED, result == STREAM_ERR_STATE ? STREAM_REASON_STOPPED : STREAM_REASON_LIMIT, now);
}
@ -1436,7 +1445,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) {
unsigned active = relay_active(net, &r->descriptor, r->peer, r->outgoing);
if (r->status < STREAM_RELAY_REJECTED && active) {
if (r->status != STREAM_RELAY_BOUND) {
r->status = STREAM_RELAY_BOUND; r->dirty = 1; r->progress_ms = now;
r->status = STREAM_RELAY_BOUND; r->dirty = r->urgent = 1; r->progress_ms = now;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay request bound to issuance source=%llx peer=%llx out=%d flows=%u",
(unsigned long long)r->descriptor.source_node_id, (unsigned long long)r->peer, r->outgoing, active);
}
@ -1463,14 +1472,15 @@ static void relay_tick(struct stream_network *net, uint64_t now) {
for (unsigned sub = 0; sub < local.descriptor.n && !available; sub++)
available = info_payload(&local, (uint8_t)sub, UINT64_MAX, r->peer, net->inst->node_id, info) == 0;
if (available) {
r->status = STREAM_RELAY_READY; r->deadline_ms = now + STREAM_RELAY_HANDOFF_MS; r->dirty = 1;
r->status = STREAM_RELAY_READY; r->deadline_ms = now + STREAM_RELAY_HANDOFF_MS;
r->dirty = r->urgent = 1;
}
}
}
}
}
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->urgent && r->sent_ms && now - r->sent_ms < 1000) continue;
struct stream_message m = {.descriptor = r->descriptor, .handle = r->id,
.sender_epoch = r->outgoing ? r->epoch : net->relay_epoch,
.receiver_epoch = r->outgoing ? r->server_epoch : r->epoch};
@ -1482,7 +1492,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) {
stream_write_be(payload + 2, remaining, 4); length = 6;
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; r->urgent = 0;
if (route_control(net, r->peer, m, payload, length, r) == STREAM_OK) r->dirty = 0;
}
struct stream_relay **link = &net->relays;

2
src/stream/stream_priv.h
Unescape View File

@ -11,7 +11,7 @@ struct stream_local {
struct stream_descriptor descriptor;
struct stream_buffer *buffer;
uint64_t start, last, next;
int anchored, publisher, finished, relay_only;
int anchored, decoded, publisher, finished, relay_only;
const uint8_t *end;
};
/* Заимствованные буфер/END действуют только до следующего callback/мутации сервиса. */

Write Preview
Loading…
Cancel
Save