diff options
| -rw-r--r-- | research/experiments/019-idle-links/transport.c | 26 | ||||
| -rw-r--r-- | serve/link.c | 87 | ||||
| -rw-r--r-- | serve/link.h | 8 |
3 files changed, 75 insertions, 46 deletions
diff --git a/research/experiments/019-idle-links/transport.c b/research/experiments/019-idle-links/transport.c index ad8e5f3..9972cd3 100644 --- a/research/experiments/019-idle-links/transport.c +++ b/research/experiments/019-idle-links/transport.c @@ -27,11 +27,24 @@ static PureModule sender, receiver; static PureLinkListener *listener; static atomic_uint_fast64_t delivered; +static unsigned char *payload; +static size_t payload_size; +static uint64_t heard, wrong; /* the last record taken; deliveries not as sent */ +/* Records as sent, each once, in order: the one after the last taken on + * (one taken before would be a delivery again), of the payload's bytes. */ static bool deliver(void *context, const PureLinkDelivery *d) { (void)context; + const unsigned char *at = d->records; + for (unsigned k = 0; k < d->count; ++k) { + uint32_t size = (uint32_t)at[0] | (uint32_t)at[1] << 8 | (uint32_t)at[2] << 16 | (uint32_t)at[3] << 24; + if (d->first + k != heard + 1 + k || size != payload_size || memcmp(at + 4, payload, size)) ++wrong; + at += 4 + size; + } + if ((size_t)(at - d->records) != d->size || d->first > heard + 1) ++wrong; + if (d->first + d->count - 1 > heard) heard = d->first + d->count - 1; atomic_fetch_add(&delivered, d->count); - pure_link_done(listener, d->token, d->first + d->count - 1, true, false); + pure_link_done(listener, d->token, heard, true, false); return true; } @@ -128,8 +141,9 @@ int main(int argc, char **argv) { if (!link) return fprintf(stderr, "transport: %s\n", error), 1; pure_link_peer(link, public_b); int output = pure_module_input(&sender, "ordered"); - unsigned char *payload = malloc(size + 256); - size_t payload_size = order(&sender, (uint32_t)output, size, payload); + payload = malloc(size + 256); + payload_size = order(&sender, (uint32_t)output, size, payload); + for (size_t i = 0; i < payload_size; ++i) payload[i] ^= (unsigned char)(i * 7); /* bytes that differ by place */ uint64_t first = 1; if (!window) /* a backlog: all queued before the link starts */ for (; first <= count; ++first) @@ -153,13 +167,13 @@ int main(int argc, char **argv) { PureLinkCounts sent; pure_link_counts(link, &sent); printf("{\"records\":%llu,\"record_bytes\":%zu,\"seconds\":%.3f,\"per_second\":%.0f,\"packets\":%llu," - "\"resent\":%llu,\"link_us_per_record\":%.3f,\"listener_us_per_record\":%.3f,\"other_us_per_record\":%.3f}\n", + "\"wrong\":%llu,\"resent\":%llu,\"link_us_per_record\":%.3f,\"listener_us_per_record\":%.3f,\"other_us_per_record\":%.3f}\n", (unsigned long long)count, payload_size, seconds, (double)count / seconds, (unsigned long long)sent.packets, - (unsigned long long)sent.resent, (double)link_ns / 1e3 / (double)count, (double)listen_ns / 1e3 / (double)count, + (unsigned long long)wrong, (unsigned long long)sent.resent, (double)link_ns / 1e3 / (double)count, (double)listen_ns / 1e3 / (double)count, (double)(all_ns - link_ns - listen_ns) / 1e3 / (double)count); pure_link_close(link); PureLinkCounts heard; pure_link_listener_close(listener, &heard); snprintf(path, sizeof(path), "rm -rf %s", dir); - return system(path) ? 0 : 0; + return system(path) || wrong ? 1 : 0; } diff --git a/serve/link.c b/serve/link.c index b8f85a7..e2da052 100644 --- a/serve/link.c +++ b/serve/link.c @@ -649,37 +649,47 @@ static size_t fill(PureLink *l, unsigned char *body, Flight *f) { * it are sent, and every one found sent moves that mark on. */ size_t k = l->count && l->unsent_from > base + 1 ? (size_t)(l->unsent_from - base - 1) : 0; bool before = true; - /* Whole records whose numbers follow on share a RECORDS frame (its count - * at run); a piece of a larger one goes alone (RECORD). */ + /* A record none of whose pieces went yet goes whole when it fits what is + * left of the packet: whole records whose numbers follow on share a + * RECORDS frame (its count at run). Else its unsent pieces go, each in a + * RECORD frame. */ size_t run = 0; uint64_t next = 0; for (; k < l->count && f->count < UNITS; ++k) { Record *r = record_at(l, k); if (r->seq > base + l->credit) break; - bool unsent = false; - for (uint32_t i = 0; i < r->pieces; ++i) { - if ((*piece_of(r, i)).state != UNSENT) continue; - uint32_t offset = i * PIECE, length = (uint32_t)least(PIECE, r->size - offset); - bool whole = r->pieces == 1, joins = whole && run && r->seq == next; - size_t need = whole ? (joins ? 0 : RECORDS_HEAD) + varint_size(length) + length : RECORD_HEAD + length; - if (f->count == UNITS || at + need > BODY) { - unsent = true; - break; + bool unsent = false, fresh = true, joins = run && r->seq == next; + for (uint32_t i = 0; i < r->pieces && fresh; ++i) fresh = piece_of(r, i)->state == UNSENT; + size_t whole = (joins ? 0 : RECORDS_HEAD) + varint_size(r->size) + r->size; + if (fresh && f->count + r->pieces <= UNITS && at + whole <= BODY) { + if (!joins) { + body[at] = F_RECORDS; + put64(body + at + 1, r->seq); + body[at + 9] = 0; + run = at + 9; + at += RECORDS_HEAD; + } + at += put_varint(body + at, r->size); + if (r->size) memcpy(body + at, l->ring + r->at, r->size); + at += r->size; + ++body[run]; + next = r->seq + 1; + for (uint32_t i = 0; i < r->pieces; ++i) { + set_piece(l, r, i, FLYING); + piece_of(r, i)->pn = f->pn; + f->units[f->count++] = (Unit){r->seq, i}; } - if (whole) { - if (!joins) { - body[at] = F_RECORDS; - put64(body + at + 1, r->seq); - body[at + 9] = 0; - run = at + 9; - at += RECORDS_HEAD; + ++l->counts.records; + } else if (fresh && r->pieces == 1) + unsent = true; /* too large for what is left: the next packet */ + else + for (uint32_t i = 0; i < r->pieces; ++i) { + if (piece_of(r, i)->state != UNSENT) continue; + uint32_t offset = i * PIECE, length = (uint32_t)least(PIECE, r->size - offset); + if (f->count == UNITS || at + RECORD_HEAD + length > BODY) { + unsent = true; + break; } - at += put_varint(body + at, length); - if (length) memcpy(body + at, l->ring + r->at, length); - at += length; - ++body[run]; - next = r->seq + 1; - } else { body[at] = F_RECORD; put64(body + at + 1, r->seq); put32(body + at + 9, r->size); @@ -688,12 +698,11 @@ static size_t fill(PureLink *l, unsigned char *body, Flight *f) { if (length) memcpy(body + at + RECORD_HEAD, l->ring + r->at + offset, length); at += RECORD_HEAD + length; run = 0; + set_piece(l, r, i, FLYING); + piece_of(r, i)->pn = f->pn; + f->units[f->count++] = (Unit){r->seq, i}; + ++l->counts.records; } - set_piece(l, r, i, FLYING); - (*piece_of(r, i)).pn = f->pn; - f->units[f->count++] = (Unit){r->seq, i}; - ++l->counts.records; - } if (before && !unsent) l->unsent_from = r->seq + 1; before = before && !unsent; if (unsent) { /* the packet is full: whole, so it can go in a burst */ @@ -1172,10 +1181,11 @@ static Held *held_at(Stream *st, size_t k) { return &st->held[st->held_first + k static uint64_t line_end(const Stream *st) { return st->line_first + st->line_count; } /* Room for more bytes at the line's end: its front let go given back - * first, else a longer line. */ + * first when that is at least half of it (so no byte moves twice in a + * doubling), else a longer line. */ static bool line_room(Stream *st, size_t more) { if (st->line_size + more <= st->line_capacity) return true; - if (st->line_head) { + if (st->line_head && st->line_head >= st->line_capacity / 2) { memmove(st->line, st->line + st->line_head, st->line_size - st->line_head); st->line_size -= st->line_head; st->line_head = 0; @@ -1454,7 +1464,7 @@ static void hold(Stream *st, uint64_t seq, uint32_t size, uint32_t offset, const } if ((st->known && seq > st->heard + CREDIT) || size > RECORD_MOST) return; if (st->known && seq < line_end(st)) return; /* on the line already */ - if (st->known && seq == line_end(st) && !offset && length == size && size <= PIECE) { /* the next, whole */ + if (st->known && seq == line_end(st) && !offset && length == size) { /* the next, whole */ if (st->held_bytes + (st->line_size - st->line_head) + size <= HELD_MOST && line_append(st, bytes, size)) line_pull(st); return; @@ -1487,10 +1497,15 @@ static void hold(Stream *st, uint64_t seq, uint32_t size, uint32_t offset, const st->held_bytes += size; } Held *h = held_at(st, k); - uint32_t piece = offset / PIECE; - if (h->size != size || offset % PIECE || piece >= h->pieces || length != least(PIECE, size - offset) || - h->bytes[size + piece]) + if (h->size != size) return; + if (!offset && length == size) { /* whole: every piece had */ + if (size) memcpy(h->bytes, bytes, size); + memset(h->bytes + size, 1, h->pieces); + h->got = h->pieces; return; + } + uint32_t piece = offset / PIECE; + if (offset % PIECE || piece >= h->pieces || length != least(PIECE, size - offset) || h->bytes[size + piece]) return; if (length) memcpy(h->bytes + offset, bytes, length); h->bytes[size + piece] = 1; ++h->got; @@ -1606,7 +1621,7 @@ static bool take_from(PureLinkListener *s, Peer *peer, unsigned char *packet, si at += RECORDS_HEAD - 1; unsigned i = 0; for (size_t used; i < count && (used = get_varint(plain + at, size - at, &length)) && - length <= PIECE && length <= size - at - used; ++i) { + length <= size - at - used; ++i) { if (st) hold(st, first + i, (uint32_t)length, 0, plain + at + used, (uint32_t)length); at += used + length; } diff --git a/serve/link.h b/serve/link.h index 5d4a746..83cc645 100644 --- a/serve/link.h +++ b/serve/link.h @@ -29,8 +29,8 @@ * shape of the messages (the sender's output and the * receiver's input must have the same): the stream the * records after it are of - * 0x02 RECORD u64 number, u32 size, u32 offset, u16 length, bytes: a saved - * send as stored, or a piece of one + * 0x02 RECORD u64 number, u32 size, u32 offset, u16 length, bytes: a piece + * of a saved send (pieces are 1000 bytes, the last shorter) * 0x03 ACK u64 largest number, u32 delay (µs), u8 ranges, each as u64 * first and u64 last, newest first * 0x04 STORED source, input (as STREAM), u64 the last number heard @@ -40,8 +40,8 @@ * 0x05 PING to be acknowledged * 0x06 RECORDS u64 the first one's number, u8 count, then whole records one * after another (numbers following on), each its size (a - * varint: LEB128) and its bytes: small sends, 1 or 2 bytes - * of framing each, not the 19 of RECORD + * varint: LEB128) and its bytes: a send that fits the packet + * whole, with 1 to 3 bytes of framing, not the 19 of RECORD * 0x00 PADDING to the end of the packet: a full packet sent in a burst is * a whole datagram * Losses are detected as RFC 9002 says; congestion control is its NewReno |