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

Measure stream issuances, probe RTT and pending data demand

master
evgeny 3 days ago
parent
921340ae60
commit
a07c53a397
1 changed files with 162 additions and 25 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. 187
      src/stream/stream_network.c

187
src/stream/stream_network.c
Unescape View File

@ -14,13 +14,20 @@
#define STREAM_FLOW_LIMIT 256
#define STREAM_PEER_FLOW_LIMIT 64
#define STREAM_CONTROL_LIMIT 128
#define STREAM_CONTROL_MAX (STREAM_WIRE_HEADER + STREAM_INFO_SIZE + 1 + STREAM_PATH_MAX * 8)
#define STREAM_CONTROL_MAX (STREAM_WIRE_HEADER + STREAM_OVERLOAD_HEADER + STREAM_OVERLOAD_MAX * 16)
#define STREAM_RENEW_TB 20000
#define STREAM_LEASE_TB 60000
#define STREAM_CONNECT_TB 50000
#define STREAM_TICK_TB 1000
struct stream_peer;
struct stream_cut { uint64_t handle, rate; int closed; };
struct stream_pressure_request {
struct stream_cut cuts[STREAM_OVERLOAD_MAX];
uint64_t id, required_rate, released_rate, deadline_ms;
unsigned count, required_count, released_count;
int active, sent;
};
struct STREAM_RECEIVE {
struct STREAM_RECEIVE *next;
struct stream_peer *peer;
@ -29,6 +36,11 @@ struct STREAM_RECEIVE {
uint64_t requested, deadline, renewed;
uint8_t subblock;
int outgoing, owned, bootstrap, info_sent, info_ready;
struct stream_metrics metrics, offered_metrics;
struct stream_load load;
uint64_t admitted_ms, pending_highest, probe, probe_sent_ms;
uint64_t gap_from;
int gap_pending, account_failed;
};
struct stream_peer {
struct stream_peer *next;
@ -40,6 +52,10 @@ struct stream_peer {
void *pump;
uint64_t node_id, own_epoch, other_epoch, next_handle, highwater, born, idle, reverse_at;
int up, busy;
struct stream_load load;
struct stream_pressure_request pressure, requested;
uint64_t pressure_id, pressure_retry_ms, clear_since_ms;
int cooling;
};
struct stream_network {
struct UTUN_INSTANCE *inst;
@ -47,12 +63,33 @@ struct stream_network {
void *timer;
unsigned peer_count, flow_count;
int stopping;
struct stream_load load;
struct uasync_load cpu;
uint64_t next_tick_ms, callback_late_ms;
int global_pressure;
};
static void peer_pump(void *arg);
static void network_tick(void *arg);
static void receive_raw(struct ETCP_CONN *conn, struct ll_entry *entry);
static void receive_routed(struct ETCP_CONN *conn, struct ll_entry *entry);
static void pressure_tick(struct stream_network *net, uint64_t now);
static void pressure_receive(struct stream_peer *peer, const struct stream_message *message);
/* Один блок принадлежит логической очереди выдачи/пира/модуля; ни одна из них не хранит копию DATA. */
static void load_account(struct STREAM_RECEIVE *f, uint64_t packets, uint64_t bytes, int added, int sent) {
if (!packets) return;
struct stream_load *loads[] = {&f->load, &f->peer->load, &f->peer->network->load};
for (unsigned i = 0; i < 3; i++) {
int result = added ? stream_load_add(loads[i], packets, bytes) : stream_load_remove(loads[i], packets, bytes, sent);
if (result != STREAM_OK) {
f->account_failed = 1;
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream ledger failed level=%u peer=%llx handle=%llu add=%d sent=%d packets=%llu bytes=%llu",
i, (unsigned long long)f->peer->node_id, (unsigned long long)f->info.handle, added, sent,
(unsigned long long)packets, (unsigned long long)bytes);
}
}
}
static int same_stream(const struct stream_descriptor *a, const struct stream_descriptor *b) {
return a->group_id == b->group_id && a->source_node_id == b->source_node_id && a->source_epoch == b->source_epoch;
@ -135,6 +172,17 @@ static int route_control(struct stream_peer *p, struct stream_message m, const u
static void flow_close(struct STREAM_RECEIVE *f, int reason, int notify) {
if (f->info.state == STREAM_RECEIVE_CLOSED) return;
struct stream_peer *p = f->peer;
if (!f->outgoing && p) {
for (unsigned i = 0; i < p->pressure.count; i++) {
struct stream_cut *cut = &p->pressure.cuts[i];
if (p->pressure.active && !cut->closed && cut->handle == f->info.handle) {
cut->closed = 1; p->pressure.released_rate += cut->rate; p->pressure.released_count++;
}
}
load_account(f, f->load.queued.packets, f->load.queued.bytes, 0, 0);
stream_load_changed(&p->load); stream_load_changed(&p->network->load);
}
stream_metrics_end(&f->metrics); stream_metrics_end(&f->offered_metrics);
if (notify && p && f->info.handle) {
uint8_t payload = (uint8_t)reason;
struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .subblock = f->subblock,
@ -176,6 +224,8 @@ static void peer_reset(struct stream_peer *p, int reason) {
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) flow_close(f, reason, 0);
p->other_epoch = p->highwater = 0; p->next_handle = 1; p->last_sent = NULL;
p->own_epoch = new_epoch(); peer_sweep(p);
memset(&p->pressure, 0, sizeof(p->pressure)); memset(&p->requested, 0, sizeof(p->requested));
p->pressure_id = 0; p->cooling = 0; stream_load_init(&p->load, get_time_tb() / 10);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer session reset peer=%llx epoch=%llx reason=%d",
(unsigned long long)p->node_id, (unsigned long long)p->own_epoch, reason);
}
@ -207,6 +257,7 @@ static struct stream_peer *peer_open(struct stream_network *net, uint64_t node)
p = u_calloc(1, sizeof(*p));
if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream peer allocation failed"); return NULL; }
p->network = net; p->node_id = node; p->own_epoch = new_epoch(); p->next_handle = 1; p->born = get_time_tb();
stream_load_init(&p->load, p->born / 10);
p->control = queue_new(net->inst->ua, 0, 0, 0, "stream_control");
if (!p->control || !p->own_epoch || node_conn_direct_open(net->inst, node, ncd_changed, p, &p->ncd, NULL) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream peer NCD/init failed peer=%llx", (unsigned long long)node);
@ -252,6 +303,9 @@ static struct STREAM_RECEIVE *flow_new(struct stream_peer *p, const struct strea
struct STREAM_RECEIVE *f = u_calloc(1, sizeof(*f));
if (!f) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream issuance allocation failed"); return NULL; }
f->peer = p; f->descriptor = *d; f->subblock = sub; f->outgoing = outgoing;
f->admitted_ms = get_time_tb() / 10; stream_load_init(&f->load, f->admitted_ms);
stream_metrics_init(&f->metrics); stream_metrics_init(&f->offered_metrics);
if (!outgoing) { stream_load_changed(&p->load); stream_load_changed(&p->network->load); }
f->info.peer = p->node_id; f->deadline = get_time_tb() + STREAM_CONNECT_TB;
f->next = p->flows; p->flows = f; p->network->flow_count++; p->idle = 0; return f;
}
@ -278,24 +332,25 @@ static int info_payload(struct stream_local *local, uint8_t sub, uint64_t seq, u
|| stream_buffer_path(local->buffer, actual, sub, &path) != STREAM_OK) return STREAM_REASON_POSITION;
if (stream_path_validate(&path, local->descriptor.source_node_id, self, peer) != STREAM_OK || path.count == STREAM_PATH_MAX)
return STREAM_REASON_LOOP;
stream_write_be(payload, actual, 8); stream_write_be(payload + 8, first, 8); stream_write_be(payload + 16, last, 8);
payload[24] = local->descriptor.n; payload[25] = local->descriptor.r;
stream_write_be(payload + 26, local->descriptor.block_size, 2);
stream_path_write(payload + 28, &path); return 0;
stream_write_be(payload + 8, actual, 8); stream_write_be(payload + 16, first, 8); stream_write_be(payload + 24, last, 8);
payload[32] = local->descriptor.n; payload[33] = local->descriptor.r;
stream_write_be(payload + 34, local->descriptor.block_size, 2);
stream_path_write(payload + STREAM_INFO_PATH_OFF, &path); return 0;
}
static int parse_info(const struct stream_message *m, const struct stream_descriptor *d, uint64_t peer, uint64_t self, uint64_t *start) {
struct stream_path path; size_t used;
if (m->length < STREAM_INFO_SIZE || m->payload[24] != d->n || m->payload[25] != d->r
|| stream_read_be(m->payload + 26, 2) != d->block_size
|| stream_path_parse(m->payload + 28, m->length - 28, &path, &used) != STREAM_OK || m->length != 28 + used
if (m->length < STREAM_INFO_SIZE || m->payload[32] != d->n || m->payload[33] != d->r
|| stream_read_be(m->payload + 34, 2) != d->block_size
|| stream_path_parse(m->payload + STREAM_INFO_PATH_OFF, m->length - STREAM_INFO_PATH_OFF, &path, &used) != STREAM_OK
|| m->length != STREAM_INFO_PATH_OFF + used
|| path.count == STREAM_PATH_MAX || stream_path_validate(&path, d->source_node_id, peer, self) != STREAM_OK
|| stream_read_be(m->payload, 8) == UINT64_MAX || stream_read_be(m->payload, 8) > stream_read_be(m->payload + 16, 8)
|| stream_read_be(m->payload + 8, 8) > stream_read_be(m->payload + 16, 8)) {
|| stream_read_be(m->payload + 8, 8) == UINT64_MAX || stream_read_be(m->payload + 8, 8) > stream_read_be(m->payload + 24, 8)
|| stream_read_be(m->payload + 16, 8) > stream_read_be(m->payload + 24, 8)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO/ACCEPT invalid command=%u bytes=%zu", m->command, m->length);
return STREAM_ERR_FORMAT;
}
*start = stream_read_be(m->payload, 8); return STREAM_OK;
*start = stream_read_be(m->payload + 8, 8); return STREAM_OK;
}
/* Один пакет на свободный порог; собственный буфер переживает синхронное освобождение entry в etcp_send. */
@ -335,6 +390,12 @@ static int downstream_send(struct STREAM_RECEIVE *f) {
struct stream_peer *p = f->peer; struct stream_local local;
if (stream_local_get(p->network->inst, &f->descriptor, &local) != STREAM_OK) { flow_close(f, STREAM_REASON_STOPPED, 1); return 0; }
struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .subblock = f->subblock};
if (f->gap_pending) {
uint8_t payload[16]; stream_write_be(payload, f->gap_from, 8); stream_write_be(payload + 8, f->info.next, 8);
m.command = STREAM_CMD_GAP;
if (raw_send(p, m, payload, sizeof(payload)) != STREAM_OK) { flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0; }
f->gap_pending = 0; f->info.gaps++; return 1;
}
const uint8_t *wire; size_t length; uint64_t actual;
int result = f->info.next == UINT64_MAX ? STREAM_WAIT
: stream_buffer_read(local.buffer, f->info.next, f->subblock, &wire, &length, &actual);
@ -348,7 +409,9 @@ static int downstream_send(struct STREAM_RECEIVE *f) {
size_t path_bytes = stream_path_write(payload, &path); memcpy(payload + path_bytes, wire, length);
m.command = STREAM_CMD_DATA;
if (raw_send(p, m, payload, path_bytes + length) != STREAM_OK) { flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0; }
if (f->info.state == STREAM_RECEIVE_ACTIVE) { f->info.next = actual + 1; f->info.blocks++; }
if (f->info.state == STREAM_RECEIVE_ACTIVE) {
load_account(f, 1, length, 0, 1); f->info.next = actual + 1; f->info.blocks++;
}
return 1;
}
if (result == STREAM_EXPIRED || result == STREAM_GAP) {
@ -383,7 +446,11 @@ static void prepare_requests(struct stream_peer *p) {
struct stream_message m = {.descriptor = f->descriptor, .subblock = f->subblock};
if (!f->info_sent) {
m.command = STREAM_CMD_INFO_REQ;
if (control_put(p, m, NULL, 0) == STREAM_OK) { f->info_sent = 1; f->info.state = STREAM_RECEIVE_INFO; }
uint8_t probe[8];
if (!f->probe) f->probe = new_epoch();
if (!f->probe) { flow_close(f, STREAM_REASON_PROTOCOL, 0); continue; }
stream_write_be(probe, f->probe, 8);
if (control_put(p, m, probe, sizeof(probe)) == STREAM_OK) { f->info_sent = 1; f->info.state = STREAM_RECEIVE_INFO; }
}
if (!f->info_ready || !p->other_epoch) continue;
uint64_t requested = f->requested;
@ -421,6 +488,12 @@ static void peer_pump(void *arg) {
int sent = 0;
struct ll_entry *e = queue_data_get(p->control);
if (e) {
if (e->len >= STREAM_WIRE_HEADER && e->dgram[2] == STREAM_CMD_INFO_REQ) {
struct stream_message m;
if (stream_wire_parse(e->dgram, e->len, &m) == STREAM_OK) {
struct STREAM_RECEIVE *f = flow_find(p, &m, 1); if (f) f->probe_sent_ms = get_time_tb() / 10;
}
}
if (etcp_send(conn, e) != 0) {
queue_dgram_free(e); queue_entry_free(e);
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream control send failed peer=%llx", (unsigned long long)p->node_id);
@ -446,7 +519,7 @@ static void peer_pump(void *arg) {
/* Известный повтор продлевает аренду без смены cursor; любой новый handle обязан превысить watermark. */
static void subscribe_request(struct stream_peer *p, struct stream_message m, struct stream_local *local) {
uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8];
uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8] = {0};
int reason = 0;
if (m.length != 8 || !m.handle || m.subblock >= local->descriptor.n) reason = STREAM_REASON_PROTOCOL;
struct STREAM_RECEIVE *f = reason ? NULL : flow_find(p, &m, 0);
@ -460,8 +533,16 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st
f = flow_new(p, &local->descriptor, m.subblock, 0);
if (!f) reason = STREAM_REASON_LIMIT;
else {
f->info.handle = m.handle; f->requested = requested; f->info.start = f->info.next = stream_read_be(payload, 8);
f->info.handle = m.handle; f->requested = requested; f->info.start = f->info.next = stream_read_be(payload + 8, 8);
f->info.state = STREAM_RECEIVE_ACTIVE;
uint64_t count = 0;
stream_buffer_pending(local->buffer, f->info.next, f->subblock, &count, NULL);
load_account(f, count, count * stream_block_size(&f->descriptor), 1, 0);
f->pending_highest = stream_read_be(payload + 24, 8);
const uint8_t *wire; size_t length; uint64_t seq;
if (stream_buffer_read(local->buffer, UINT64_MAX, f->subblock, &wire, &length, &seq) == STREAM_OK)
stream_metrics_observe(&f->offered_metrics, seq, (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2),
wire[STREAM_BLOCK_FLAGS_OFF], (uint32_t)length, get_time_tb() / 10);
}
}
} else if (f && !reason) {
@ -469,7 +550,7 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st
/* Идемпотентный ACCEPT возвращает первоначальный индекс даже если тот уже вытеснен. */
if (reason == STREAM_REASON_POSITION || reason == STREAM_REASON_EXPIRED) {
reason = info_payload(local, m.subblock, UINT64_MAX, p->node_id, p->network->inst->node_id, payload);
if (!reason) stream_write_be(payload, f->info.start, 8);
if (!reason) stream_write_be(payload + 8, f->info.start, 8);
}
}
if (reason) {
@ -483,8 +564,11 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st
return;
}
f->deadline = get_time_tb() + STREAM_LEASE_TB;
struct stream_metric_snapshot snapshot = {0};
int valid = stream_local_metrics(p->network->inst, &local->descriptor, m.subblock, get_time_tb() / 10, &snapshot) == STREAM_OK;
stream_metric_write(payload + STREAM_INFO_METRIC_OFF, &snapshot, valid);
m.command = STREAM_CMD_ACCEPT;
if (control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8) != STREAM_OK) {
if (control_put(p, m, payload, STREAM_INFO_SIZE + payload[STREAM_INFO_PATH_OFF] * 8) != STREAM_OK) {
flow_close(f, STREAM_REASON_LIMIT, 0); return;
}
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream ACCEPT queued peer=%llx handle=%llu sub=%u start=%llu lease=6000ms",
@ -495,6 +579,13 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st
static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t *wire, size_t length, int routed) {
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_OVERLOAD || m.command == STREAM_CMD_OVERLOAD_SELECT) {
struct stream_peer *p = peer_find(net, sender);
if (!routed && p && p->up && m.receiver_epoch == p->own_epoch && m.sender_epoch == p->other_epoch && !m.handle) {
p->busy++; pressure_receive(p, &m); p->busy--; peer_sweep(p);
} else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/unknown pressure request peer=%llx", (unsigned long long)sender);
return;
}
if (stream_local_get(net->inst, &m.descriptor, &local) != STREAM_OK) {
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream packet ignored without local subscription command=%u", m.command); return;
}
@ -514,7 +605,7 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed DATA/control forbidden command=%u", m.command); return;
}
if (m.command == STREAM_CMD_INFO_REQ && !routed) {
if (m.handle || m.length || m.subblock >= local.descriptor.n) {
if (m.handle || m.length != 8 || !stream_read_be(m.payload, 8) || m.subblock >= local.descriptor.n) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO_REQ malformed peer=%llx bytes=%zu handle=%llu sub=%u n=%u",
(unsigned long long)sender, m.length, (unsigned long long)m.handle, m.subblock, local.descriptor.n); return;
}
@ -526,11 +617,18 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
/* Remote INFO начинает новое поколение; его следующий ответ подтвердит свежий local epoch. */
}
p->other_epoch = m.sender_epoch;
uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8];
uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8] = {0};
memcpy(payload, m.payload, 8);
int reason = info_payload(&local, m.subblock, UINT64_MAX, sender, net->inst->node_id, payload);
m.command = reason ? STREAM_CMD_REJECT : STREAM_CMD_INFO_RESP;
if (reason) { uint8_t value = (uint8_t)reason; control_put(p, m, &value, 1); }
else control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8);
if (reason) {
uint8_t failure[9] = {(uint8_t)reason}; memcpy(failure + 1, m.payload, 8); control_put(p, m, failure, sizeof(failure));
} else {
struct stream_metric_snapshot snapshot = {0};
int valid = stream_local_metrics(net->inst, &local.descriptor, m.subblock, get_time_tb() / 10, &snapshot) == STREAM_OK;
stream_metric_write(payload + STREAM_INFO_METRIC_OFF, &snapshot, valid);
control_put(p, m, payload, STREAM_INFO_SIZE + payload[STREAM_INFO_PATH_OFF] * 8);
}
return;
}
if (!p || (!p->up && !routed) || m.receiver_epoch != p->own_epoch
@ -545,8 +643,13 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
&& m.command != STREAM_CMD_UNSUBSCRIBE);
if (m.command == STREAM_CMD_INFO_RESP && f && !m.handle && f->info_sent) {
uint64_t start;
if (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &start) == STREAM_OK) {
if (m.length < 8 || stream_read_be(m.payload, 8) != f->probe) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale INFO probe peer=%llx", (unsigned long long)sender);
} else if (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &start) == STREAM_OK
&& stream_metric_read(m.payload + STREAM_INFO_METRIC_OFF, &f->info.supplier_metrics, &f->info.supplier_metrics_valid) == STREAM_OK) {
p->other_epoch = m.sender_epoch; f->info_ready = 1; peer_schedule(p);
uint64_t rtt = get_time_tb() / 10 - f->probe_sent_ms;
f->info.rtt_valid = f->probe_sent_ms != 0; f->info.rtt_ms = rtt > UINT32_MAX ? UINT32_MAX : (uint32_t)rtt;
} else flow_close(f, STREAM_REASON_PROTOCOL, 0);
} else if (m.command == STREAM_CMD_SUBSCRIBE) {
if (!authorized(net, sender, &m, &local)) {
@ -575,6 +678,8 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
else {
int result = stream_ingest_path(net->inst, block, m.length - used, &path);
if (result == STREAM_OK || result == STREAM_DUPLICATE || result == STREAM_EXPIRED) {
stream_metrics_observe(&f->metrics, seq, (uint16_t)stream_read_be(block + STREAM_BLOCK_TIMESTAMP_OFF, 2),
block[STREAM_BLOCK_FLAGS_OFF], (uint32_t)stream_block_size(&f->descriptor), get_time_tb() / 10);
f->info.next = seq + 1; f->info.blocks++; f->deadline = get_time_tb() + STREAM_LEASE_TB;
} else flow_close(f, STREAM_REASON_PROTOCOL, 1);
}
@ -586,7 +691,9 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
} else if (m.command == STREAM_CMD_END && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
uint64_t last = m.length == STREAM_END_SIZE ? stream_read_be(m.payload + 34, 8) : UINT64_MAX;
if (m.length == STREAM_END_SIZE && last != UINT64_MAX && f->info.next == last + 1
&& stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) f->info.state = STREAM_RECEIVE_ENDED;
&& stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) {
f->info.state = STREAM_RECEIVE_ENDED; stream_metrics_end(&f->metrics);
}
else flow_close(f, STREAM_REASON_PROTOCOL, 1);
} else if (m.command == STREAM_CMD_RENEW && f && !m.length) {
f->deadline = get_time_tb() + STREAM_LEASE_TB;
@ -594,7 +701,9 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
(unsigned long long)sender, (unsigned long long)m.handle, (unsigned long long)f->deadline);
}
else if (m.command == STREAM_CMD_UNSUBSCRIBE && f && m.length == 1) flow_close(f, m.payload[0], 1);
else if ((m.command == STREAM_CMD_CLOSED || m.command == STREAM_CMD_REJECT) && f && (m.length == 1 || m.length == 17))
else if (m.command == STREAM_CMD_REJECT && f && !m.handle && m.length == 9 && stream_read_be(m.payload + 1, 8) == f->probe)
flow_close(f, m.payload[0], 0);
else if ((m.command == STREAM_CMD_CLOSED || m.command == STREAM_CMD_REJECT) && f && m.handle && (m.length == 1 || m.length == 17))
flow_close(f, m.payload[0], 0);
else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream unknown/inactive issuance peer=%llx command=%u handle=%llu sub=%u bytes=%zu",
(unsigned long long)sender, m.command, (unsigned long long)m.handle, m.subblock, m.length);
@ -636,6 +745,9 @@ static void transport_changed(struct ETCP_CONN *conn, int status, void *arg) {
static void network_tick(void *arg) {
struct stream_network *net = arg; net->timer = NULL;
uint64_t now = get_time_tb();
uint64_t now_ms = now / 10;
net->callback_late_ms = now_ms > net->next_tick_ms ? now_ms - net->next_tick_ms : 0;
uasync_get_load(net->inst->ua, &net->cpu);
struct stream_peer *p = net->peers;
while (p) {
struct stream_peer *next = p->next;
@ -660,6 +772,8 @@ static void network_tick(void *arg) {
if (p->idle && now - p->idle >= STREAM_LEASE_TB && !p->control->count) peer_free(p);
p = next;
}
pressure_tick(net, now_ms);
net->next_tick_ms = get_time_tb() / 10 + STREAM_TICK_TB / 10;
net->timer = uasync_set_timeout(net->inst->ua, STREAM_TICK_TB, net, network_tick, "stream_network");
if (!net->timer) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream network timer scheduling failed");
}
@ -672,6 +786,7 @@ int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out)
struct stream_network *net = u_calloc(1, sizeof(*net));
if (!net) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream network allocation failed"); return STREAM_ERR_MEMORY; }
net->inst = inst;
stream_load_init(&net->load, get_time_tb() / 10); net->next_tick_ms = get_time_tb() / 10 + STREAM_TICK_TB / 10;
if (etcp_bind(inst, ETCP_ID_STREAM, receive_raw) < 0 || etcp_router_bind(inst, ETCP_RT_ID_STREAM, receive_routed) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream bindings registration failed"); stream_network_destroy(net); return STREAM_ERR_STATE;
}
@ -789,5 +904,27 @@ int stream_receive_restart(struct STREAM_HANDLE *subscription) {
int stream_receive_get_info(const struct STREAM_RECEIVE *f, struct stream_receive_info *info) {
if (!f || !info) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream receive info invalid arguments"); return STREAM_ERR_ARGUMENT; }
*info = f->info; return STREAM_OK;
*info = f->info;
info->metrics_valid = stream_metrics_poll((struct stream_metrics *)&f->metrics, get_time_tb() / 10, &info->metrics) == STREAM_OK;
if (f->peer && f->peer->requested.active) for (unsigned i = 0; i < f->peer->requested.count; i++)
if (f->peer->requested.cuts[i].handle == f->info.handle) info->close_requested = 1;
return STREAM_OK;
}
/* Добавление/вытеснение учитывает только ещё не переданные блоки действующей downstream-выдачи. */
void stream_network_cache_event(struct stream_network *net, const struct stream_descriptor *d, const uint8_t *wire, size_t length, int added) {
if (!net || net->stopping) return;
uint64_t seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8); uint8_t sub = wire[STREAM_BLOCK_SUB_OFF];
for (struct stream_peer *p = net->peers; p; p = p->next) {
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
if (f->outgoing || f->info.state != STREAM_RECEIVE_ACTIVE || f->subblock != sub || seq < f->info.next
|| !same_stream(&f->descriptor, d)) continue;
load_account(f, 1, length, added, 0);
if (added) {
if (seq > f->pending_highest) f->pending_highest = seq;
stream_metrics_observe(&f->offered_metrics, seq, (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2),
wire[STREAM_BLOCK_FLAGS_OFF], (uint32_t)length, get_time_tb() / 10);
}
}
}
}

Write Preview
Loading…
Cancel
Save