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

Retain stream block paths and locate gaps in ring delivery

master
evgeny 4 days ago
parent
18fe51ad25
commit
b31aa9f71d
3 changed files with 55 additions and 1 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. 2
      src/Makefile.am
  2. 48
      src/stream/stream_buffer.c
  3. 6
      src/stream/stream_buffer.h

2
src/Makefile.am
Unescape View File

@ -7,6 +7,8 @@ stream_core_files = \
stream/stream.h \
stream/stream.c \
stream/stream_priv.h \
stream/stream_wire.h \
stream/stream_wire.c \
stream/stream_member.h \
stream/stream_member.c \
stream/stream_proto.h \

48
src/stream/stream_buffer.c
Unescape View File

@ -9,6 +9,7 @@
struct stream_cached_slot {
uint32_t next;
uint8_t subblock;
struct stream_path *path; /* NULL — исходный путь {source}, без отдельной аллокации. */
};
struct stream_cached_batch {
@ -102,6 +103,7 @@ static void buffer_evict_oldest(struct stream_buffer *b) {
for (uint32_t slot = batch->first_slot; slot != STREAM_SLOT_NONE;) {
uint32_t next = b->slots[slot].next;
unsigned sub = b->slots[slot].subblock;
u_free(b->slots[slot].path); b->slots[slot].path = NULL;
if (b->latest_valid[sub] && b->latest[sub] == batch->seq) b->latest_valid[sub] = 0;
b->slots[slot].next = b->free_slot;
b->free_slot = slot;
@ -169,6 +171,7 @@ void stream_buffer_destroy(struct stream_buffer *b) {
(unsigned long long)b->descriptor.source_node_id, b->stats.blocks, b->stats.batches, b->stats.stored_bytes,
(unsigned long long)b->stats.evicted_batches, (unsigned long long)b->stats.duplicates);
for (uint32_t i = 0; i < b->stats.batches; i++) u_free(b->heap[i]);
if (b->slots) for (uint32_t i = 0; i < b->capacity; i++) u_free(b->slots[i].path);
rs_free(b->rs);
u_free(b->decode_output);
u_free(b->data);
@ -180,7 +183,17 @@ void stream_buffer_destroy(struct stream_buffer *b) {
/* Вставить только подписанный блок; OOM/конфликт проверяются до вытеснения ранее сохранённых данных. */
int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t length) {
struct stream_path path = {.count = 1};
if (b) path.nodes[0] = b->descriptor.source_node_id;
return stream_buffer_put_path(b, wire, length, &path);
}
/* Путь и память готовятся до вытеснения: отказ не повреждает окно кольца. */
int stream_buffer_put_path(struct stream_buffer *b, const uint8_t *wire, size_t length, const struct stream_path *path) {
if (!b || !wire) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer put invalid arguments length=%zu", length); return STREAM_ERR_ARGUMENT; }
if (!path || !path->count || path->count > STREAM_PATH_MAX
|| stream_path_validate(path, b->descriptor.source_node_id, path->nodes[path->count - 1], 0) != STREAM_OK)
return STREAM_ERR_FORMAT;
struct stream_cached_batch *batch = NULL;
if (length == b->block_bytes) {
batch = buffer_batch(b, stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8));
@ -214,12 +227,18 @@ int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t lengt
(unsigned long long)view.seq, view.subblock, (unsigned long long)b->heap[0]->seq);
return STREAM_EXPIRED;
}
struct stream_path *stored_path = path->count > 1 ? u_malloc(sizeof(*stored_path)) : NULL;
if (path->count > 1 && !stored_path) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer path allocation failed seq=%llu", (unsigned long long)view.seq);
return STREAM_ERR_MEMORY;
}
if (stored_path) *stored_path = *path;
int new_batch = batch == NULL;
if (new_batch) {
batch = u_calloc(1, sizeof(*batch));
if (!batch) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer batch allocation failed seq=%llu", (unsigned long long)view.seq);
return STREAM_ERR_MEMORY;
u_free(stored_path); return STREAM_ERR_MEMORY;
}
batch->seq = view.seq;
batch->first_slot = STREAM_SLOT_NONE;
@ -232,6 +251,7 @@ int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t lengt
uint32_t slot = b->free_slot;
b->free_slot = b->slots[slot].next;
b->slots[slot].subblock = view.subblock;
b->slots[slot].path = stored_path;
b->slots[slot].next = batch->first_slot;
batch->first_slot = slot;
batch->count++;
@ -248,6 +268,32 @@ int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t lengt
return STREAM_OK;
}
/* Копия пути не держит указатели в переиспользуемом кольце. */
int stream_buffer_path(struct stream_buffer *b, uint64_t seq, uint8_t subblock, struct stream_path *path) {
if (!b || !path || subblock >= b->descriptor.n || seq == UINT64_MAX) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer path invalid arguments"); return STREAM_ERR_ARGUMENT;
}
struct stream_cached_batch *batch = buffer_batch(b, seq);
uint32_t slot = batch ? buffer_slot(b, batch, subblock) : STREAM_SLOT_NONE;
if (slot == STREAM_SLOT_NONE) return seq < b->stats.expired_before ? STREAM_EXPIRED : STREAM_GAP;
*path = b->slots[slot].path ? *b->slots[slot].path : (struct stream_path){.nodes = {b->descriptor.source_node_id}, .count = 1};
return STREAM_OK;
}
/* Поиск используется только на дыре/вытеснении, обычная выдача читает точный индекс через hash. */
int stream_buffer_next_seq(struct stream_buffer *b, uint64_t seq, uint8_t subblock, uint64_t *next) {
if (!b || !next || subblock >= b->descriptor.n) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer next invalid arguments"); return STREAM_ERR_ARGUMENT;
}
uint64_t found = UINT64_MAX;
for (uint32_t i = 0; i < b->stats.batches; i++) {
struct stream_cached_batch *batch = b->heap[i];
if (batch->seq >= seq && batch->seq < found && buffer_slot(b, batch, subblock) != STREAM_SLOT_NONE) found = batch->seq;
}
if (found == UINT64_MAX) return STREAM_WAIT;
*next = found; return STREAM_OK;
}
/* Найти точную позицию либо последний блок субпотока, не меняя его историю/срок хранения. */
int stream_buffer_read(struct stream_buffer *b, uint64_t seq, uint8_t subblock,
const uint8_t **wire, size_t *length, uint64_t *actual_seq) {

6
src/stream/stream_buffer.h
Unescape View File

@ -7,6 +7,7 @@
#define STREAM_BUFFER_H
#include "stream_block.h"
#include "stream_wire.h"
#ifdef __cplusplus
extern "C" {
@ -34,6 +35,11 @@ void stream_buffer_destroy(struct stream_buffer *buffer);
/* Новый блок проверяется до изменения кольца. Повтор неизменного блока: STREAM_DUPLICATE.
* STREAM_EXPIRED не меняет более свежие данные; неверная подпись/конфликт возвращает отрицательный код. */
int stream_buffer_put(struct stream_buffer *buffer, const uint8_t *wire, size_t length);
/* Сохранённая история содержит текущего владельца кольца; первый путь повтора остаётся неизменным. */
int stream_buffer_put_path(struct stream_buffer *buffer, const uint8_t *wire, size_t length, const struct stream_path *path);
int stream_buffer_path(struct stream_buffer *buffer, uint64_t seq, uint8_t subblock, struct stream_path *path);
/* Найти ближайший имеющийся блок >= seq для скачка через GAP; WAIT при отсутствии. */
int stream_buffer_next_seq(struct stream_buffer *buffer, uint64_t seq, uint8_t subblock, uint64_t *next);
/* seq=UINT64_MAX читает последний имеющийся блок именно запрошенного субпотока.
* WAIT — блок впереди головы/пустое кольцо; GAP — дыра; EXPIRED — вытеснен.

Write Preview
Loading…
Cancel
Save