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

Add internal ownership for exact-epoch stream relays

master
evgeny 6 days ago
parent
8883002e3f
commit
cd5e658fb9
2 changed files with 71 additions and 18 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. 81
      src/stream/stream.c
  2. 8
      src/stream/stream_priv.h

81
src/stream/stream.c
Unescape View File

@ -32,9 +32,9 @@ struct stream_session {
uint64_t restart_from, restart_last;
uint8_t end[STREAM_END_SIZE];
int end_valid;
unsigned owners, holds;
unsigned owners, relay_owners, holds;
int anchored, closing, invalidated, finished, end_delivered, reason;
int restarting, restart_gap;
int restarting, restart_gap, notify_close;
};
struct STREAM_HANDLE {
@ -42,6 +42,7 @@ struct STREAM_HANDLE {
struct stream_service *service;
struct stream_session *session;
int publisher;
int relay;
int selection_owned;
int reason;
};
@ -196,7 +197,8 @@ static int service_publish_options(struct stream_service *service, struct TOPO_G
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "local subscription list exceeds member limit group=%llx",
(unsigned long long)group->group_id); return STREAM_ERR_ARGUMENT;
}
struct stream_subscription_key entry = {s->descriptor.source_node_id, s->descriptor.source_epoch};
struct stream_subscription_key entry = {s->descriptor.source_node_id, s->descriptor.source_epoch,
!s->source && s->owners == s->relay_owners};
size_t at = state.count++;
while (at && (state.subscriptions[at - 1].source_node_id > entry.source_node_id
|| (state.subscriptions[at - 1].source_node_id == entry.source_node_id
@ -251,6 +253,7 @@ static void session_schedule(struct stream_session *s) {
/* Снять локальные ссылки и публикацию; CLOSED также отложен и отменяется при stop всего сервиса. */
static void session_close(struct stream_session *s, int reason, int update_profile, int notify) {
if (s->closing) return;
s->notify_close = notify && s->owners > s->relay_owners;
s->closing = 1; s->reason = reason; s->owners = 0;
stream_optimizer_destroy(s->optimizer); s->optimizer = NULL;
stream_network_cancel(s->service->network, &s->descriptor);
@ -279,7 +282,7 @@ static void session_deliver(void *arg) {
s->delivery = NULL; s->holds++;
struct stream_event event = { .descriptor = s->descriptor };
int emit = 0, again = 0;
if (s->closing) { event.type = STREAM_EVENT_CLOSED; event.reason = s->reason; emit = 1; }
if (s->closing) { event.type = STREAM_EVENT_CLOSED; event.reason = s->reason; emit = s->notify_close; }
else if (s->invalidated) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "invalidated stream awaits deferred close source=%llx",
(unsigned long long)s->descriptor.source_node_id);
@ -316,6 +319,7 @@ static void session_deliver(void *arg) {
event.type = STREAM_EVENT_END; event.first_seq = event.last_seq = UINT64_MAX;
s->end_delivered = 1; emit = 1;
}
if (!s->closing && s->owners == s->relay_owners) emit = 0;
if (emit && service->callback && !service->stopping) {
service->dispatching++;
service->callback(service->inst, &event, service->arg);
@ -493,11 +497,12 @@ void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callb
/* Подготовить все ресурсы до публикации member-опций; rollback не оставляет видимую подписку. */
static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, const uint8_t key[32], int publisher, int automatic,
int relay,
struct STREAM_HANDLE **out) {
struct stream_service *service = inst->stream;
struct stream_session *s = session_find(service, d->group_id, d->source_node_id);
int created = s == NULL;
if (s && (publisher || s->descriptor.source_epoch != d->source_epoch || (!s->source && !!s->optimizer != automatic))) {
if (s && (publisher || s->descriptor.source_epoch != d->source_epoch || (!relay && !s->source && !!s->optimizer != automatic))) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream instance conflicts with publication/epoch/selection mode automatic=%d existing_auto=%d",
automatic, s->optimizer != NULL); return STREAM_ERR_STATE;
}
@ -523,10 +528,12 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript
for (unsigned i = 0; i < d->n; i++) stream_metrics_init(&s->metrics[i]);
stream_buffer_set_observer(s->buffer, cache_event, s);
}
h->service = service; h->session = s; h->publisher = publisher;
h->next = service->handles; service->handles = h; s->owners++;
int was_relay = s->owners && s->owners == s->relay_owners;
h->service = service; h->session = s; h->publisher = publisher; h->relay = relay;
h->next = service->handles; service->handles = h; s->owners++; s->relay_owners += relay;
s->holds++;
int result = created ? service_publish_options(service, stream_group(inst, d->group_id)) : STREAM_OK;
int result = created || was_relay != (s->owners == s->relay_owners)
? service_publish_options(service, stream_group(inst, d->group_id)) : STREAM_OK;
if (result == STREAM_OK && created && utun_set_client_activity(inst, inst->client_activity_requested) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream open failed to activate core"); result = STREAM_ERR_STATE;
}
@ -537,7 +544,10 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript
struct STREAM_HANDLE **link = &service->handles;
while (*link != h) link = &(*link)->next;
*link = h->next;
if (!s->closing) { s->owners--; session_close(s, STREAM_CLOSE_FAILED, 0, 0); }
if (!s->closing) {
s->owners--; s->relay_owners -= relay;
if (created) session_close(s, STREAM_CLOSE_FAILED, 0, 0);
}
u_free(h); s->holds--;
if (!s->holds && !s->delivery) session_free(s);
return result;
@ -577,14 +587,14 @@ int stream_publish_open(struct UTUN_INSTANCE *inst, uint64_t group_id, const str
}
} while (!d.source_epoch);
if (stream_descriptor_validate(&d) != STREAM_OK) return STREAM_ERR_ARGUMENT;
return session_open(inst, &d, inst->my_ed25519_pubkey, 1, 0, out);
return session_open(inst, &d, inst->my_ed25519_pubkey, 1, 0, 0, out);
}
static int subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, int automatic, struct STREAM_HANDLE **out) {
int result = stream_open_check(inst, group_id, out);
if (result != STREAM_OK) return result;
struct stream_session *s = session_find(inst->stream, group_id, source_id);
if (s) return session_open(inst, &s->descriptor, s->public_key, 0, automatic, out);
if (s) return session_open(inst, &s->descriptor, s->public_key, 0, automatic, 0, out);
struct stream_member_state state; uint8_t key[32];
result = stream_member_load(inst, group_id, source_id, &state, key);
if (result != STREAM_OK) return result;
@ -592,7 +602,25 @@ static int subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source not yet published source=%llx", (unsigned long long)source_id);
return STREAM_WAIT;
}
return session_open(inst, &state.source, key, 0, automatic, out);
return session_open(inst, &state.source, key, 0, automatic, 0, out);
}
/* Запрос старой эпохи не может незаметно подписать суперузел на новый запуск. */
int stream_relay_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct STREAM_HANDLE **out) {
int result = key ? stream_open_check(inst, key->group_id, out) : STREAM_ERR_ARGUMENT;
if (result != STREAM_OK) return result;
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 session_open(inst, &s->descriptor, s->public_key, 0, 1, 1, out);
struct stream_member_state state; uint8_t public_key[32];
result = stream_member_load(inst, key->group_id, key->source_node_id, &state, public_key);
if (result != STREAM_OK) return result;
if (!state.published || state.source.source_epoch != key->source_epoch) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay source epoch unavailable source=%llx requested=%llx published=%d actual=%llx",
(unsigned long long)key->source_node_id, (unsigned long long)key->source_epoch, state.published,
(unsigned long long)state.source.source_epoch); return STREAM_ERR_STATE;
}
return session_open(inst, &state.source, public_key, 0, 1, 1, out);
}
int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out) {
@ -627,7 +655,18 @@ void stream_close(struct STREAM_HANDLE *h) {
*link = h->next;
}
struct stream_session *s = h->session;
if (s && (h->publisher || --s->owners == 0)) session_close(s, STREAM_CLOSE_STOPPED, 1, 1);
if (s) {
int was_relay = s->owners == s->relay_owners;
if (h->publisher || s->owners == 1) session_close(s, STREAM_CLOSE_STOPPED, 1, !h->relay);
else {
s->owners--; s->relay_owners -= h->relay;
if (was_relay != (s->owners == s->relay_owners)
&& service_publish_options(service, stream_group(service->inst, s->descriptor.group_id)) != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream owner mode publication failed; closing inconsistent instance");
session_close(s, STREAM_CLOSE_FAILED, 0, 1);
}
}
}
u_free(h);
}
@ -762,7 +801,8 @@ 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, .end = s->end_valid ? s->end : NULL};
.anchored = s->anchored, .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;
}
@ -862,13 +902,20 @@ int stream_get_candidates(const struct STREAM_HANDLE *h, struct stream_candidate
int source = node->node_id == s->descriptor.source_node_id;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream candidate peer=%llx type=%u source=%d online=1",
(unsigned long long)node->node_id, identity ? identity->client_type : UINT8_MAX, source);
if (identity && identity->client_type == CLIENT_TYPE_MOBILE && !source) {
if (identity && identity->client_type == CLIENT_TYPE_MOBILE && !source
&& stream_member_supernode(inst, group->group_id, node->node_id) != 1) {
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream candidate excluded peer=%llx reason=mobile_relay",
(unsigned long long)node->node_id); continue;
}
struct stream_member_state state;
if (stream_member_load(inst, group->group_id, node->node_id, &state, NULL) != STREAM_OK
|| !stream_member_has_subscription(&state, s->descriptor.source_node_id, s->descriptor.source_epoch)) continue;
if (stream_member_load(inst, group->group_id, node->node_id, &state, NULL) != STREAM_OK) continue;
int confirmed = stream_relay_confirmed(inst, &s->descriptor, node->node_id);
if (!stream_member_has_subscription(&state, s->descriptor.source_node_id, s->descriptor.source_epoch) && !confirmed) continue;
int relay_only = confirmed;
for (size_t i = 0; i < state.count; i++)
if (state.subscriptions[i].source_node_id == s->descriptor.source_node_id
&& state.subscriptions[i].source_epoch == s->descriptor.source_epoch) relay_only = state.subscriptions[i].relay_only;
if (!source && s->owners == s->relay_owners && relay_only) continue;
candidates[found++] = (struct stream_candidate){ .node_id = node->node_id, .source = (uint8_t)source,
.client_type = identity ? identity->client_type : UINT8_MAX };
}

8
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;
int anchored, publisher, finished, relay_only;
const uint8_t *end;
};
/* Заимствованные буфер/END действуют только до следующего callback/мутации сервиса. */
@ -25,6 +25,12 @@ void stream_local_progress(struct UTUN_INSTANCE *inst, const struct stream_descr
int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key);
int stream_local_metrics(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint8_t sub, uint64_t now_ms,
struct stream_metric_snapshot *snapshot);
/* Внутреннее владение точной эпохой; close обязателен и после detach. Режим существующего экземпляра сохраняется. */
int stream_relay_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct STREAM_HANDLE **out);
/* Подтверждение поставщика действует только в живой попытке помощи, до прихода Merkle-опции. */
int stream_relay_confirmed(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t peer);
/* Вызывается selector один раз на экземпляр; relay-only экземпляры новых запросов помощи не создают. */
void stream_relay_demand(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, int needed, uint64_t now_ms);
int stream_ingest_block(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length);

Write Preview
Loading…
Cancel
Save