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

Probe all online attachment sources and choose the lowest measured RTT

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 13:02:20 +01:00
1 parent c5c6d96384
commit bfd14abdbb
1 file changed
+131 -4
Download Patch File Download Diff File Expand all files Collapse all files
src/dm/dm_media.c
+131 -4
View File
@@ -30,11 +30,13 @@
enum dm_file_work { DM_FILE_PREPARE, DM_FILE_RECEIVE, DM_FILE_CUSTODY, DM_FILE_DELETE };
struct dm_file_job;
struct dm_file_probe;
struct dm_media_state {
struct UTUN_INSTANCE* inst;
char root[768];
char pending[768];
struct dm_file_job* jobs;
struct dm_file_probe* probes;
void* timer;
int closing;
int cleanup_active;
@@ -58,6 +60,17 @@ struct dm_file_job {
unsigned held;
};
#define DM_SOURCE_MAX (DM_TARGET_MAX + 64)
struct dm_probe_source { uint64_t node, group, sent, rtt; int used; };
struct dm_file_probe {
struct dm_file_probe* next;
uint8_t desc[DM_MEDIA_DESC_SIZE];
uint64_t token, deadline, author;
char conv[64];
unsigned count;
struct dm_probe_source sources[DM_SOURCE_MAX];
};
sqlite3_stmt* dm_media_prepare(struct UTUN_INSTANCE* inst, const char* sql) {
sqlite3_stmt* st = NULL;
int rc = sqlite3_prepare_v2(inst->topo_sqlite_db, sql, -1, &st, NULL);
@@ -741,7 +754,7 @@ static void dm_file_pull(struct dm_media_state* media, int custody, uint64_t gro
}
}
/* Общий авторизатор сервера transfer: A отдаёт B/storage, M — только B по custody. */
/* Поставщиком может быть любая готовая реплика; UUID сам по себе права не даёт. */
static int dm_file_lookup(void* arg, uint64_t group, uint64_t peer, const uint8_t id[16], uint64_t size,
char* path, size_t cap) {
struct dm_media_state* media = arg;
@@ -749,8 +762,8 @@ static int dm_file_lookup(void* arg, uint64_t group, uint64_t peer, const uint8_
if (media->closing) return -1;
for (int custody = 0; custody <= 1; custody++) {
sqlite3_stmt* st = dm_media_prepare(inst, custody ?
"SELECT author,recipient,plain_size,ready,expires FROM dm_custody WHERE id=?" :
"SELECT author,recipient,plain_size,ready,0 FROM dm_files WHERE id=? AND dir=1");
"SELECT author,recipient,plain_size,ready,expires,body,NULL FROM dm_custody WHERE id=?" :
"SELECT author,recipient,plain_size,ready,0,body,conv_id FROM dm_files WHERE id=?");
if (!st) return -1;
sqlite3_bind_blob(st, 1, id, 16, SQLITE_STATIC);
int rc = sqlite3_step(st);
@@ -759,10 +772,17 @@ static int dm_file_lookup(void* arg, uint64_t group, uint64_t peer, const uint8_
uint64_t plain = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 2) : 0;
int ready = rc == SQLITE_ROW ? sqlite3_column_int(st, 3) : 0;
uint64_t expiry = rc == SQLITE_ROW ? (uint64_t)sqlite3_column_int64(st, 4) : 0;
int allowed = 0;
struct dm_message_view view;
if (rc == SQLITE_ROW && !dm_message_parse(sqlite3_column_blob(st, 5), (size_t)sqlite3_column_bytes(st, 5), &view)) {
if (view.mode == DM_MODE_DEVICE) allowed = peer == author || peer == recipient;
else allowed = dm_party_member(inst, view.parties, peer) == 1 || dm_party_member(inst, view.parties + 1, peer) == 1;
}
if (!custody && rc == SQLITE_ROW && dm_conversation_access(inst, (const char*)sqlite3_column_text(st, 6), NULL, NULL)) ready = 0;
sqlite3_finalize(st);
if (!author || ready != 1 || size != dm_media_cipher_size(plain) ||
(custody && expiry <= ntp_time_get_seconds(inst))) continue;
if ((peer == recipient && dm_media_scope(inst, group, peer, NULL, author, recipient)) ||
if ((allowed && dm_media_scope(inst, group, peer, NULL, inst->node_id, peer)) ||
(!custody && dm_media_scope(inst, group, peer, "storage", author, recipient)))
return dm_file_path(media, custody, id, ".cipher", path, cap);
}
@@ -770,6 +790,109 @@ static int dm_file_lookup(void* arg, uint64_t group, uint64_t peer, const uint8_
return -1;
}
/* Не менять активную передачу из-за позднего ответа; оставшиеся ответы — резерв. */
static struct dm_file_probe* dm_probe_find(struct dm_media_state* media, const uint8_t id[16]) {
for (struct dm_file_probe* p = media->probes; p; p = p->next) if (!memcmp(p->desc, id, 16)) return p;
return NULL;
}
static void dm_probe_add(struct dm_media_state* media, struct dm_file_probe* p, uint64_t node, uint64_t preferred) {
if (node == media->inst->node_id) return;
for (unsigned i = 0; i < p->count; i++) if (p->sources[i].node == node) return;
uint64_t group;
if (!dm_route_group(media->inst, node, preferred, &group)) return;
if (p->count == DM_SOURCE_MAX) { DEBUG_WARN(DEBUG_CATEGORY_DM, "DM sources: candidate limit reached conv=%s", p->conv); return; }
struct dm_probe_source* source = p->sources + p->count++;
source->node = node; source->group = group; source->sent = get_time_tb();
uint8_t request[64]; uint64_t plain, size; memcpy(&plain, p->desc + 16, 8); size = dm_media_cipher_size(plain);
memcpy(request, p->desc, 16); memcpy(request + 16, p->desc + 24, 32);
memcpy(request + 48, &p->token, 8); memcpy(request + 56, &size, 8);
if (dm_media_send(media->inst, group, node, DM_MEDIA_HAVE, request, sizeof(request))) source->used = 1;
}
/* Одна ограниченная параллельная проба по устройствам обеих сторон и известным M. */
static void dm_probe_start(struct dm_media_state* media, const uint8_t* body, size_t len, uint64_t preferred) {
struct dm_message_view view;
if (dm_message_parse(body, len, &view) || !view.media || dm_probe_find(media, view.media)) return;
struct dm_file_probe* p = u_calloc(1, sizeof(*p));
if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_DM, "DM sources: allocation failed"); return; }
memcpy(p->desc, view.media, sizeof(p->desc)); p->author = view.author;
snprintf(p->conv, sizeof(p->conv), "%llu", (unsigned long long)view.conv);
struct dm_party parties[2]; struct dm_target targets[DM_TARGET_MAX]; size_t count;
if (dm_conversation_access(media->inst, p->conv, parties, NULL) ||
dm_identity_targets(media->inst, parties, targets, DM_TARGET_MAX, &count) || RAND_bytes((uint8_t*)&p->token, 8) != 1) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "DM sources: party/source snapshot unavailable conv=%s", p->conv); u_free(p); return;
}
p->deadline = get_time_tb() + 10000; p->next = media->probes; media->probes = p;
for (size_t i = 0; i < count; i++) dm_probe_add(media, p, targets[i].node, preferred);
sqlite3_stmt* st = dm_media_prepare(media->inst, "SELECT node,group_id FROM dm_file_sources WHERE id=? AND (expires=0 OR expires>?)");
if (st) {
sqlite3_bind_blob(st, 1, p->desc, 16, SQLITE_STATIC); sqlite3_bind_int64(st, 2, ntp_time_get_seconds(media->inst));
uint64_t nodes[64], groups[64]; unsigned n = 0; int rc;
while ((rc = sqlite3_step(st)) == SQLITE_ROW && n < 64) {
nodes[n] = (uint64_t)sqlite3_column_int64(st, 0); groups[n++] = (uint64_t)sqlite3_column_int64(st, 1);
}
sqlite3_finalize(st);
if (rc != SQLITE_DONE) DEBUG_WARN(DEBUG_CATEGORY_DM, "DM sources: holder scan incomplete rc=%d count=%u", rc, n);
for (unsigned i = 0; i < n; i++) dm_probe_add(media, p, nodes[i], groups[i]);
}
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM sources: parallel HAVE conv=%s candidates=%u window_ms=1000", p->conv, p->count);
}
static void dm_probe_control(struct dm_media_state* media, uint64_t group, uint64_t peer, int reply, const uint8_t* request) {
uint64_t token, size; memcpy(&token, request + 48, 8); memcpy(&size, request + 56, 8);
if (reply) {
struct dm_file_probe* p = dm_probe_find(media, request);
if (!p || p->token != token || memcmp(request + 16, p->desc + 24, 32) || get_time_tb() > p->deadline) return;
for (unsigned i = 0; i < p->count; i++) {
struct dm_probe_source* s = p->sources + i;
if (s->node == peer && s->group == group && !s->used && !s->rtt) {
s->rtt = get_time_tb() - s->sent + 1;
DEBUG_DEBUG(DEBUG_CATEGORY_DM, "DM sources: HAVE peer=%llu rtt_us=%llu", (unsigned long long)peer,
(unsigned long long)(s->rtt * 100)); return;
}
}
return;
}
sqlite3_stmt* st = dm_media_prepare(media->inst,
"SELECT body FROM dm_files WHERE id=? UNION ALL SELECT body FROM dm_custody WHERE id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, request, 16, SQLITE_STATIC); sqlite3_bind_blob(st, 2, request, 16, SQLITE_STATIC);
int found = 0, rc;
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
const uint8_t* desc = dm_message_media(sqlite3_column_blob(st, 0), (size_t)sqlite3_column_bytes(st, 0));
if (desc && !memcmp(desc + 24, request + 16, 32)) { found = 1; break; }
}
sqlite3_finalize(st);
if (!found) return;
char path[1024]; struct stat sb;
if (!dm_file_lookup(media, group, peer, request, size, path, sizeof(path)) && !stat(path, &sb) && (uint64_t)sb.st_size == size)
dm_media_send(media->inst, group, peer, DM_MEDIA_HAVE_ACK, request, 64);
}
/* После окна проб выбираем минимальный RTT. Ошибка transfer/проверки оставляет резервные источники. */
static void dm_probe_tick(struct dm_media_state* media) {
struct dm_file_probe** link = &media->probes;
while (*link) {
struct dm_file_probe* p = *link;
if (get_time_tb() < p->deadline || dm_job_find(media, p->desc, 0)) { link = &p->next; continue; }
sqlite3_stmt* st = dm_media_prepare(media->inst, "SELECT f.ready,c.closed FROM dm_files f JOIN dm_conversations c USING(conv_id) WHERE f.id=?");
if (!st) return;
sqlite3_bind_blob(st, 1, p->desc, 16, SQLITE_STATIC); int rc = sqlite3_step(st);
int pending = rc == SQLITE_ROW && sqlite3_column_int(st, 0) != 1 && !sqlite3_column_int(st, 1);
sqlite3_finalize(st);
unsigned best = p->count;
if (pending) for (unsigned i = 0; i < p->count; i++) if (!p->sources[i].used && p->sources[i].rtt &&
(best == p->count || p->sources[i].rtt < p->sources[best].rtt)) best = i;
if (best == p->count) { *link = p->next; u_free(p); continue; }
struct dm_probe_source* s = p->sources + best; s->used = 1;
DEBUG_INFO(DEBUG_CATEGORY_DM, "DM sources: download conv=%s peer=%llu rtt_us=%llu", p->conv,
(unsigned long long)s->node, (unsigned long long)(s->rtt * 100));
dm_file_pull(media, 0, s->group, s->node, p->desc, p->author);
link = &p->next;
}
}
/* Путь исходника копируется; caller может освободить строку после возврата. */
int dm_send_file(struct UTUN_INSTANCE* inst, const char* conv, const char* path, const struct attachment_info* info,
int remove_source, uint64_t request_id) {
@@ -1087,6 +1210,8 @@ static void dm_media_control(struct UTUN_INSTANCE* inst, uint64_t group, uint64_
uint8_t cmd, const uint8_t* p, size_t len) {
if (!inst->dm_media || inst->dm_media->closing) return;
if (cmd == DM_MEDIA_OFFER) dm_media_offer(inst, group, source, p, len);
else if ((cmd == DM_MEDIA_HAVE || cmd == DM_MEDIA_HAVE_ACK) && len == 64)
dm_probe_control(inst->dm_media, group, source, cmd == DM_MEDIA_HAVE_ACK, p);
else if (cmd == DM_MEDIA_STORE) dm_custody_store(inst, group, source, p, len);
else if (cmd == DM_MEDIA_DELETE && len == DM_MEDIA_RECEIPT_SIZE) dm_custody_delete(inst, group, source, p);
else if (cmd == DM_MEDIA_RECEIPT && len == DM_MEDIA_RECEIPT_SIZE) {
@@ -1228,6 +1353,7 @@ static void dm_media_tick(void* arg) {
media->timer = NULL;
uint64_t now = ntp_time_get_seconds(media->inst);
dm_files_tick(media, now);
dm_probe_tick(media);
dm_custody_tick(media, now);
dm_cleanup_pump(media);
media->timer = uasync_set_timeout(media->inst->ua, 10000, media, dm_media_tick, "dm_media");
@@ -1283,6 +1409,7 @@ void dm_media_quiesce(struct UTUN_INSTANCE* inst) {
media->timer = NULL;
etcp_router_unbind(inst, ETCP_RT_ID_DM_MEDIA);
file_transfer_destroy(inst->file_transfer);
while (media->probes) { struct dm_file_probe* p = media->probes; media->probes = p->next; u_free(p); }
DEBUG_INFO(DEBUG_CATEGORY_DM, "dm_media: quiesced, waiting for workers");
}