Separate stream routing from member files and scope relay activation
This commit is contained in:
evgeny committed
1 parent
8dbfb4d297
commit
57d2828496
4 files changed
+126
-17
No files matched your search
@@ -32,14 +32,15 @@ struct stream_relay_request {
|
||||
struct stream_relay_request *next;
|
||||
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;
|
||||
int outgoing, status, reason, dirty, granted;
|
||||
uint8_t end[STREAM_END_SIZE];
|
||||
};
|
||||
struct stream_relay {
|
||||
struct stream_relay *next;
|
||||
struct stream_descriptor descriptor;
|
||||
struct STREAM_HANDLE *owner;
|
||||
uint64_t idle_ms;
|
||||
uint64_t idle_ms, checked_ms;
|
||||
int allowed;
|
||||
};
|
||||
|
||||
enum stream_peer_reset_flags { STREAM_RESET_ROTATE = 1, STREAM_RESET_KEEP_PENDING = 2 };
|
||||
@@ -219,6 +220,10 @@ static int route_control(struct stream_network *net, uint64_t peer, struct strea
|
||||
if (length) memcpy(e->dgram + 1 + STREAM_WIRE_HEADER, payload, length);
|
||||
e->len = 1 + STREAM_WIRE_HEADER + length;
|
||||
int result = etcp_route_send(inst, m.descriptor.group_id, peer, e, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
|
||||
if (m.command >= STREAM_CMD_RELAY_REQUEST) DEBUG_DEBUG(DEBUG_CATEGORY_STREAM,
|
||||
"relay control queued self=%llx peer=%llx group=%llx command=%u id=%llx sender=%llx receiver=%llx result=%d",
|
||||
(unsigned long long)inst->node_id, (unsigned long long)peer, (unsigned long long)m.descriptor.group_id, m.command,
|
||||
(unsigned long long)m.handle, (unsigned long long)m.sender_epoch, (unsigned long long)m.receiver_epoch, result);
|
||||
if (result < 0) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed control failed peer=%llx command=%u result=%d",
|
||||
(unsigned long long)peer, m.command, result);
|
||||
return result < 0 ? STREAM_ERR_STATE : STREAM_OK;
|
||||
@@ -743,6 +748,8 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
|
||||
struct stream_message m; struct stream_local local;
|
||||
if (stream_wire_parse(wire, length, &m) != STREAM_OK || !sender || sender == net->inst->node_id) return;
|
||||
if (m.command >= STREAM_CMD_RELAY_REQUEST) {
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay dispatcher self=%llx sender=%llx command=%u routed=%d bytes=%zu",
|
||||
(unsigned long long)net->inst->node_id, (unsigned long long)sender, m.command, routed, length);
|
||||
if (routed) relay_dispatch(net, sender, &m);
|
||||
else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay command requires signed routed control peer=%llx command=%u",
|
||||
(unsigned long long)sender, m.command);
|
||||
@@ -903,6 +910,11 @@ static void receive_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
|
||||
/* Router сообщает реального отправителя и признаки проверенной подписи, не next hop. */
|
||||
static void receive_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
|
||||
struct stream_network *net = conn ? stream_network_get(conn->instance) : NULL;
|
||||
if (entry && entry->dgram && entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
|
||||
&& entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] >= STREAM_CMD_RELAY_REQUEST)
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "relay routed entry self=%llx net=%p bytes=%u flags=%u cmd=%u",
|
||||
(unsigned long long)(conn && conn->instance ? conn->instance->node_id : 0), (void *)net, entry->len,
|
||||
entry->dgram[ROUTER_SVC_FLAGS_OFF], entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2]);
|
||||
if (net && entry && entry->dgram && entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
|
||||
&& (entry->dgram[ROUTER_SVC_FLAGS_OFF] & ROUTER_FLAG_SIGNED)) {
|
||||
uint64_t sender, group; memcpy(&sender, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
|
||||
@@ -1016,6 +1028,8 @@ int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out)
|
||||
stream_network_destroy(net); return STREAM_ERR_MEMORY;
|
||||
}
|
||||
*out = net;
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed handler installed self=%llx callback=%p net=%p",
|
||||
(unsigned long long)inst->node_id, (void *)receive_routed, (void *)net);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream network initialized raw=%u routed=%u peers=%u flows=%u controls=%u lease=6000ms renew=2000ms",
|
||||
ETCP_ID_STREAM, ETCP_RT_ID_STREAM, STREAM_PEER_LIMIT, STREAM_FLOW_LIMIT, STREAM_CONTROL_LIMIT);
|
||||
return STREAM_OK;
|
||||
@@ -1343,6 +1357,10 @@ static void relay_dispatch(struct stream_network *net, uint64_t sender, const st
|
||||
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay request nonce conflicts peer=%llx id=%llx", (unsigned long long)sender,
|
||||
(unsigned long long)m->handle); return;
|
||||
}
|
||||
if (!r && m->command == STREAM_CMD_RELAY_REQUEST && m->receiver_epoch) {
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "relay expired request confirmation ignored peer=%llx id=%llx",
|
||||
(unsigned long long)sender, (unsigned long long)m->handle); return;
|
||||
}
|
||||
if (!r) r = relay_request_new(net, &m->descriptor, sender, m->handle, m->sender_epoch, 0);
|
||||
if (!r) return;
|
||||
if (m->command == STREAM_CMD_RELAY_CANCEL) relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now);
|
||||
@@ -1359,8 +1377,11 @@ static void relay_dispatch(struct stream_network *net, uint64_t sender, const st
|
||||
&& member.subscriptions[i].source_epoch == m->descriptor.source_epoch && member.subscriptions[i].relay_only) {
|
||||
relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_LOOP, now); return;
|
||||
}
|
||||
/* Первый REQUEST только подтверждает поколение сервиса. Старый пакет после tombstone не запускает приём. */
|
||||
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 && result != STREAM_WAIT)
|
||||
relay_finish(net, r, STREAM_RELAY_REJECTED, result == STREAM_ERR_STATE ? STREAM_REASON_STOPPED : STREAM_REASON_LIMIT, now);
|
||||
}
|
||||
@@ -1396,7 +1417,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) {
|
||||
(unsigned long long)r->id, r->status);
|
||||
relay_finish(net, r, r->outgoing ? STREAM_RELAY_CANCELLED : STREAM_RELAY_REJECTED, STREAM_REASON_TRANSPORT, now);
|
||||
}
|
||||
if (!r->outgoing && r->status == STREAM_RELAY_STARTING) {
|
||||
if (!r->outgoing && r->granted && r->status == STREAM_RELAY_STARTING) {
|
||||
struct stream_local local;
|
||||
if (stream_local_get(net->inst, &r->descriptor, &local) == STREAM_OK) {
|
||||
if (local.finished && local.last == UINT64_MAX) {
|
||||
@@ -1437,8 +1458,11 @@ static void relay_tick(struct stream_network *net, uint64_t now) {
|
||||
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);
|
||||
int valid = stream_handle_active(relay->owner)
|
||||
&& stream_member_supernode(net->inst, relay->descriptor.group_id, net->inst->node_id) == 1;
|
||||
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;
|
||||
}
|
||||
int valid = stream_handle_active(relay->owner) && relay->allowed;
|
||||
if (valid && (pending || active)) { relay->idle_ms = 0; link = &relay->next; continue; }
|
||||
if (valid && !relay->idle_ms) relay->idle_ms = now;
|
||||
if (valid && now - relay->idle_ms < STREAM_RELAY_IDLE_MS) { link = &relay->next; continue; }
|
||||
|
||||
@@ -43,7 +43,8 @@ extern "C" {
|
||||
// router: 0x00 DATA, 0x02 NAT, 0x04 TCP_PROXY_SERVER, 0x05 UDP_PROXY,
|
||||
// 0x06 ICMP_PROXY, 0x07 MEDIA_DELIVERY, 0x08 TCP_PROXY_CLIENT,
|
||||
// 0x11 CONN_MGR, 0x32 ADMIN_KEY, 0x33 JOIN, 0x34 DM,
|
||||
// 0x35 DM_MAILBOX, 0x36 CALL, 0x37 FILE_TRANSFER, 0x38 DM_MEDIA, 0x39 STREAM, 0x3A GROUP_AVATAR, 0x3B NAT_DETECTION
|
||||
// 0x35 DM_MAILBOX, 0x36 CALL, 0x37 FILE_TRANSFER, 0x38 DM_MEDIA, 0x39 MEMBER_FILES,
|
||||
// 0x3A GROUP_AVATAR, 0x3B NAT_DETECTION, 0x3C STREAM
|
||||
//
|
||||
// !!! Новый ID: впиши его в нужный реестр выше и убедись, что значение
|
||||
// не занято в том же пространстве.
|
||||
@@ -58,7 +59,7 @@ extern "C" {
|
||||
#define ETCP_ID_BROADCAST 0x05 // broadcast-сообщения между пирами
|
||||
#define ETCP_ID_RADIO 0x06 // рация (PTT мультикаст голоса) между пирами
|
||||
#define ETCP_ID_STREAM 0x07 // прямой stream INFO/выдачи/DATA
|
||||
#define ETCP_RT_ID_STREAM 0x39 // stream REVERSE_REQ/CLOSED через группу
|
||||
#define ETCP_RT_ID_STREAM 0x3C // stream REVERSE_REQ/CLOSED и помощь суперузлов через группу
|
||||
|
||||
// === Router-сервисы (etcp_router_bind/etcp_route_send) ===
|
||||
#define ETCP_RT_ID_DATA 0x00 // routing.c — маршрутизация данных
|
||||
|
||||
Reference in new issue
Block a user