Size a receiver's credit by what it holds, not 1,024 records

A receiver gave every stream a credit of 1,024 records past its last
stored, however small the records: a link carried at most 1,024 records
a round trip. On a path of 100 ms that is 10,000 records a second of
61 B, a ninth of what 50 Mbit/s carries (the path simulator: 9,953 a
second, 90,814 at 10 ms); and records of 1 KiB, a queue's worth of them
in flight, went a record or two a packet as each STORED let a few more
through (M6: 107,968 packets where 20 ms took 84,106).

A receiver bounds what it holds of a stream (16 MiB) apart, so its
credit is now what fits there at the size its last delivery's records
had: 1 to 65,536 (what a sender takes), 1,024 before any delivery. It
takes records that far past its last stored, as its memory lets it.
Simulator at 100 ms: 56,296 records a second of 61 B. M6 at 100 ms:
45.4 Mbit/s of 1 KiB records in 83,566 packets (44.9 in 107,968). The
window, not the credit, now fills the bottleneck's queue: losses there
where the credit hid them (61 B at 10 ms: 87,130 a second, 1,371 lost).
012 says so.

2 files changed, 22 insertions, 9 deletions

diff --git a/docs/design/012-link-protocol.md b/docs/design/012-link-protocol.md
index c99bdb0..1ad3739 100644
--- a/docs/design/012-link-protocol.md
+++ b/docs/design/012-link-protocol.md
@@ -216,8 +216,11 @@ STORED의 들은 번호 이하인 기록은 덜어 낸다. 보내는 앱은 그�
   - 그 run이 마지막으로 보낸 스트림마다 STORED
   - REFUSED
 - STORED는 저장이 끝났을 때에도 보낸다.
-  - 허가는 1024다. 받는 쪽이 그 스트림에서 16 MiB를 쥐고 있으면 1이다.
-  - 들은 번호 + 1024보다 큰 기록은 버린다.
+  - 허가는 받는 쪽이 정한다(지역 선택). 이 구현은 그 스트림에 남은 자리(16 MiB에서 쥔 바이트를 뺀 것)를 지난 전달의 기록당
+    바이트로 나눈다. 1에서 65,536 사이이고, 아직 전달이 없으면 1,024다. 그래서 링크가 한 RTT에 나르는 기록 수가 기록의 크기를
+    따라간다.
+  - 들은 번호 + 65,536보다 큰 기록은 버린다. 그 안의 기록도 메모리(16 MiB)를 넘으면 받지 않는다. 보내는 쪽은 저장 소식이 없으니
+    다시 보낸다.
   - 들은 번호 이하의 기록이 다시 오면 STORED를 다시 보낸다.
 - 받는 쪽은 들은 번호 다음부터 번호 순서로 온전한 기록을 앱에 넘긴다. 앱이 적용하고 저장한 뒤에만 STORED의 들은 번호를 올린다.
   틈에서 멈추고, 틈 뒤의 기록을 버렸으면 깃발 1을 세운다.
@@ -271,12 +274,13 @@ LOST는 그 패킷의 번호를 담아 에폭 0의 키로 봉인한다. 답 run�
 | 첫 데이터그램 / 가장 큰 데이터그램 | 1,200 B / 1,452 B | 명세 |
 | 이름 | 1–64 B | 명세 |
 | 기록 | 16 MiB 이하 | 명세 |
-| 허가의 상한(보내는 쪽이 따르는) | 65,536 | 명세 |
+| 허가의 상한(보내는 쪽이 따르고, 받는 쪽이 받는 범위) | 65,536 | 명세 |
 | ACCEPT의 되풀이 | 4 B | 명세 |
 | 첫 허가(STORED 전) | 1,024 | 명세 |
 | 새 주소의 증폭 한도 | 받은 바이트의 3배 | 명세 |
 | LOST 크기 조건 | 부른 패킷 > 46 B | 명세 |
 | 에폭 주기 | 1시간 또는 1 GiB 이하 | 상한만 명세 |
+| STORED의 허가 | 남은 자리 ÷ 기록당 바이트(1–65,536, 처음 1,024) | 지역 선택 |
 | 거절 뒤 기다림 | 1초부터 두 배, 64초까지 | 지역 선택 |
 | 에폭을 잃었다고 볼 때 | 2초와 PTO 4번 | 지역 선택 |
 | 받은 뒤 저장을 기다리는 시간 | 1초 | 지역 선택 |
diff --git a/link/link-protocol.c b/link/link-protocol.c
index 318441e..570ab15 100644
--- a/link/link-protocol.c
+++ b/link/link-protocol.c
@@ -26,7 +26,13 @@ enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PIN
  * (a sender may offer an epoch again with another key). */
 enum { EPOCH_MASK = 15, EPOCH_FRAME = 1 + 4 + PURE_SEAL_KEY, ANSWERS = 4, ACCEPT_FRAME = EPOCH_FRAME + ANSWERS };
 enum { RECORDS_HEAD = 10 }; /* type, the first number, the count */
-enum { RANGES = 32, UNITS = 64, CREDIT = 1024, BATCH = 256, RECORD_HEAD = 19 };
+enum { RANGES = 32, UNITS = 64, BATCH = 256, RECORD_HEAD = 19 };
+/* Credit: records past the last stored a sender may send, CREDIT before a
+ * receiver says (its STORED), and what it says after, CREDIT_MOST at most
+ * (a sender takes no more): what fits in what it holds (HELD_MOST) at the
+ * size its records have had. A receiver takes records that far past the
+ * last it stored, as its memory lets it. */
+enum { CREDIT = 1024, CREDIT_MOST = 65536 };
 /* A receiver keeps a peer's runs in RUNS places (its sender's, and the one
  * before a restart), and more beside when more of them send at once (a
  * node's outputs to this one, each a sender of its own), RUNS_MOST in all. */
@@ -702,7 +708,7 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui
             for (uint32_t i = 0; i < r->parts; ++i)
                 if (part_of(r, i)->state == RECEIVED) set_part(l, r, part_of(r, i), UNSENT);
         }
-    l->limit = most(l->limit, heard + (credit ? least(credit, 65536) : 1));
+    l->limit = most(l->limit, heard + (credit ? least(credit, CREDIT_MOST) : 1));
 }
 
 /* The receiver refuses the stream: no records go for a while, then a
@@ -1274,6 +1280,7 @@ typedef struct Stream {
     unsigned line_count;
     unsigned out_count; /* the delivery out from the line: its records and bytes (let go at once when all applied) */
     size_t out_bytes;
+    uint32_t mean; /* the bytes a record took on the line in the last delivery from it (0: none yet) */
     bool tail;
     uint32_t tail_size, tail_got;
     Held *held;                  /* records not yet stored, by number: held_count from held_first */
@@ -1972,7 +1979,7 @@ static void hold(Stream *st, uint64_t seq, uint32_t size, uint32_t offset, const
         st->stored_due = true; /* sent again: say again how far it is stored */
         return;
     }
-    if ((st->known && seq > st->heard + CREDIT) || size > RECORD_MOST) return;
+    if ((st->known && seq > st->heard + CREDIT_MOST) || size > RECORD_MOST) return;
     if (st->known && seq < line_end(st)) return; /* on the line already */
     if ((!length && offset) || offset > size || length > size - offset) return;
     if (st->known && seq == line_end(st) && on_line(st, size, offset, bytes, length)) return;
@@ -2329,7 +2336,7 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack
             unsigned i = 0;
             /* In order, nothing held and no tail waiting: straight onto the
              * line, the frame at once (what hold does a record, for all). */
-            bool straight = st && st->known && first == line_end(st) && first + count - 1 <= st->heard + CREDIT &&
+            bool straight = st && st->known && first == line_end(st) && first + count - 1 <= st->heard + CREDIT_MOST &&
                             !st->tail && !st->held_count &&
                             st->line_size - st->line_head + (size - at) <= HELD_MOST &&
                             line_room(st, size - at + 4 * (size_t)count);
@@ -2434,8 +2441,9 @@ static void answer(PureLinkReceiver *s, uint64_t now, PureLinkSend *send, void *
             at += put_names(body + at, st->source, st->input_name);
             put64(body + at, st->heard);
             body[at + 8] = st->gap ? STORED_GAP : 0;
-            size_t room = st->held_bytes + (st->line_size - st->line_head) < HELD_MOST ? CREDIT : 1;
-            put32(body + at + 9, (uint32_t)room);
+            size_t used = st->held_bytes + (st->line_size - st->line_head);
+            size_t credit = used >= HELD_MOST ? 1 : st->mean ? (HELD_MOST - used) / st->mean : CREDIT;
+            put32(body + at + 9, (uint32_t)(credit < 1 ? 1 : credit > CREDIT_MOST ? CREDIT_MOST : credit));
             at += 13;
             st->stored_due = st->gap = false;
         }
@@ -2470,6 +2478,7 @@ static void apply_done(PureLinkReceiver *s, const PureLinkDone *done, size_t cou
         if (!done[k].stored) continue; /* not saved: handed over again */
         if (!st->known || done[k].heard > st->heard) st->heard = done[k].heard;
         if (!st->known) st->line_first = st->heard + 1; /* the line begins */
+        if (st->out_count) st->mean = (uint32_t)(st->out_bytes / st->out_count);
         if (st->out_count && st->heard == st->line_first + st->out_count - 1) { /* the delivery applied whole */
             st->line_head += st->out_bytes;
             st->line_first += st->out_count;