TCP message server · срез 0.0.27

Следующий прикладной срез после TCP echo добавляет границы сообщений. Соединения по-прежнему блокирующие, обслуживается один клиент за раз. Новый синтаксис, runtime-примитивы, reactor и маршрутизация между клиентами не нужны.

Сервер возвращает каждое полностью принятое сообщение в том же формате:

Поле	Формат
Длина	Ровно 4 байта, unsigned `u32`, big-endian
Тело	Ровно N произвольных байтов

Длина не включает заголовок. В приложении 0 ≤ N ≤ 65536. Пустое сообщение допустимо и получает ответ из четырёх нулевых байтов. UTF-8, команды и JSON не подразумеваются. Протокол не имеет negotiation, authentication или TLS.

Запуск и клиент

python -m liza build examples/apps/liza_message.liza -O 2 -o build/liza-message
./build/liza-message

По умолчанию адрес — 127.0.0.1:9001. --ephemeral выбирает свободный порт, --once выбирает свободный порт и нормально завершает сервер после одного клиента. stdout содержит выбранный порт и LF. На Windows используется .exe. Telnet не формирует этот бинарный заголовок; для проверки нужен framed client:

import socket
import struct

def receive_exact(stream, length):
    result = bytearray()
    while len(result) < length:
        part = stream.recv(length - len(result))
        if not part:
            raise EOFError("Incomplete frame")
        result.extend(part)
    return bytes(result)

with socket.create_connection(("127.0.0.1", 9001), timeout=5) as stream:
    payload = b"Hello Liza\x00\xff"
    stream.sendall(struct.pack("!I", len(payload)) + payload)
    length = struct.unpack("!I", receive_exact(stream, 4))[0]
    assert length <= 65536
    assert receive_exact(stream, length) == payload
    stream.shutdown(socket.SHUT_WR)

Клиент должен читать ответы по мере отправки сообщений: бесконечная отправка без чтения может заполнить TCP-буферы и заблокировать обе стороны.

Библиотека framing

liza/stdlib/framing.liza — чистый код Liza. Парсер не выделяет память, не копирует payload, не хранит view и не зависит от времени жизни входного буфера. Он возвращает индексы сегмента payload, который приложение должно обработать до переиспользования входных байтов.

Операция	Контракт
`framing.create(limit)`	Независимый `FrameState`, готовый к заголовку
`framing.feed(state, data, start, count)`	`Result<FrameStep,FrameError>` для диапазона `View<u8>`
`framing.finish(state)`	`Ok(0)` только на границе кадров; иначе ошибка усечённого входа
`framing.header(length)`	Четыре байта big-endian для любого `u32`

FrameState — копируемое значение с четырьмя u32: лимит, накопленная длина, число байтов заголовка и число принятых байтов тела. Поля приватны. feed возвращает новую версию состояния; исходная версия остаётся независимой. Лимит библиотеки может быть любым u32, но сервер фиксирует 64 КиБ.

FrameStep содержит:

Поле	Смысл
`state`	Следующее состояние; после завершения кадра готово к следующему заголовку
`next`	Абсолютный индекс первого ещё не обработанного байта текущего view
`payload_start`, `payload_count`	Доступный сегмент тела в текущем view
`body_offset`	Позиция этого сегмента в теле сообщения
`length`	Длина тела после полного заголовка; до его завершения равна нулю
`complete`	Полностью получен один кадр, включая случай N = 0

Один feed завершается не позже конца текущего кадра. Если next меньше конца входного диапазона, вызывающий код продолжает с возвращённым состоянием и оставшимися байтами. Поэтому за одно чтение могут завершиться несколько кадров.

Для непустого допустимого входа успешный вызов всегда продвигает next. Пустой диапазон не создаёт кадр и сохраняет состояние. Проверки диапазона предшествуют чтению и используют count <= data.len-start, исключая переполнение суммы индексов. За один вызов читается не более четырёх байтов заголовка; payload описывается диапазоном без сканирования его содержимого.

FrameError:

InvalidRange — передан недопустимый диапазон view.
TooLarge — полный заголовок объявляет длину больше лимита.
TruncatedHeader — EOF после 1–3 байтов нового заголовка.
TruncatedBody — EOF после полного заголовка, до завершения тела.

Ошибка — Result, не trap. При Err новая версия состояния не возвращается; сервер завершает соединение, не пытаясь найти следующий заголовок в недостоверном потоке. Независимый вызывающий код может начать новый разбор через create.

Сборка сообщения и стоимость

На соединение создаётся один регион ёмкостью 67584 байта: 2048 для чтения и 65536 для тела. Оба буфера инициализируются один раз. Они переиспользуются до закрытия соединения; per-message аллокаций и reserve нет. Служебные значения и заголовок имеют фиксированный размер. Память сетевого стека ОС — отдельно.

Сегмент тела копируется из входного буфера в соответствующий диапазон буфера сообщения. Всего это N байтов на сообщение; весь буфер не копируется на каждом чтении. Передача тела по сети использует readonly view и точную длину N. Байты за пределами N не отправляются, в том числе после более длинного сообщения. Это не zero-copy: хранение полного кадра нужно, чтобы не выдавать ответ на усечённое сообщение.

До полного кадра ответ не начинается. Затем отправляются заголовок и тело, каждый с циклом досылки. Успешный net.write непустого диапазона гарантирует положительный прогресс; I/O-ошибка завершает соединение. Ошибка отправки уже полного кадра может оставить у клиента частичный ответ — сетевой вывод не атомарен и не откатывается.

EOF на границе кадров — нормальное завершение. При ошибке последующего кадра предыдущие ответы не отменяются; это не гарантия их доставки приложению peer. Сервер не обещает сохранить соединение после ошибки протокола.

Исходник

examples/apps/liza_message.liza
// Wire format: u32 big-endian length followed by that many arbitrary bytes.
// Only complete frames are echoed. The configured maximum is 65536 bytes.
fn exchange(stream:net.Stream, initial:memory.Buffer<u8>, storage:memory.Buffer<u8>)->u32:io {
    var input=move initial
    var body=move storage
    var state=framing.create(65536)
    var running=true
    var status:u32=0
    while running {
        (next,received)=net.read(stream,move input)
        input=move next
        match received {
            Err(error)=>{status=3;running=false}
            Ok(count)=>{
                if count==0 {
                    match framing.finish(state) {Ok(done)=>{} Err(error)=>{status=4}}
                    running=false
                } else {
                    data=memory.view(input)
                    var cursor:u32=0
                    while cursor<count && running {
                        match framing.feed(state,data,cursor,count-cursor) {
                            Err(error)=>{status=4;running=false}
                            Ok(step)=>{
                                state=step.state
                                cursor=step.next
                                var i:u32=0
                                while i<step.payload_count {
                                    body=memory.set(move body,step.body_offset+i,data[step.payload_start+i])
                                    i=i+1
                                }
                                if step.complete {
                                    header=framing.header(step.length)
                                    prefix=view header[0..4]
                                    var sent:u32=0
                                    while sent<4 && running {
                                        match net.write(stream,prefix,sent,4-sent) {
                                            Err(error)=>{status=3;running=false}
                                            Ok(n)=>{sent=sent+n}
                                        }
                                    }
                                    payload=memory.view(body)
                                    sent=0
                                    while sent<step.length && running {
                                        match net.write(stream,payload,sent,step.length-sent) {
                                            Err(error)=>{status=3;running=false}
                                            Ok(n)=>{sent=sent+n}
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }
    released_input=memory.release(move input)
    released_body=memory.release(move body)
    match net.close_stream(move stream) {Ok(done)=>{} Err(error)=>{status=3}}
    return status
}
fn serve(stream:net.Stream)->u32:io,alloc {
    match memory.create(67584) {
        Err(error)=>{closed=net.close_stream(move stream);return 3}
        Ok(region)=>{
            match memory.alloc<u8>(region,2048,0) {
                Err(error)=>{
                    freed=memory.close(move region)
                    closed=net.close_stream(move stream)
                    return 3
                }
                Ok(input)=>{
                    match memory.alloc<u8>(region,65536,0) {
                        Err(error)=>{
                            released=memory.release(move input)
                            freed=memory.close(move region)
                            closed=net.close_stream(move stream)
                            return 3
                        }
                        Ok(body)=>{
                            status=exchange(move stream,move input,move body)
                            freed=memory.close(move region)
                            return status
                        }
                    }
                }
            }
        }
    }
}
fn main()->u32:io,alloc {
    var once=false
    var port:u16=9001
    if cli.count()!=1 {
        if cli.count()!=2 {message=console.err("Usage: liza-message [--once | --ephemeral]\n");return 2}
        match cli.arg(1) {
            None=>{return 2}
            Some(arg)=>{
                if !cli.equals(arg,"--once") && !cli.equals(arg,"--ephemeral") {message=console.err("Usage: liza-message [--once | --ephemeral]\n");return 2}
                once=cli.equals(arg,"--once")
                port=0
            }
        }
    }
    match net.listen(port) {
        Err(error)=>{message=console.err("Cannot listen on loopback\n");return 3}
        Ok(listener)=>{
            var status:u32=0
            var running=true
            match net.local_port(listener) {
                Err(error)=>{status=3;running=false}
                Ok(actual)=>{
                    match console.print_u64(u64(actual)) {Ok(done)=>{} Err(error)=>{status=3;running=false}}
                    match console.write("\n") {Ok(done)=>{} Err(error)=>{status=3;running=false}}
                }
            }
            while running {
                match net.accept(listener) {
                    Err(error)=>{status=3;running=false}
                    Ok(stream)=>{
                        result=serve(move stream)
                        if result!=0 {
                            status=result
                            if result==4 {
                                match console.err("Invalid frame\n") {Ok(done)=>{} Err(error)=>{status=3;running=false}}
                            } else {
                                match console.err("Connection I/O failed\n") {Ok(done)=>{} Err(error)=>{status=3;running=false}}
                            }
                        }
                        if once {running=false}
                    }
                }
            }
            match net.close_listener(move listener) {Ok(done)=>{} Err(error)=>{status=3}}
            return status
        }
    }
}

exchange передаёт владение входным буфером в net.read и получает его обратно через линейное произведение. View входа живёт только в обработке одного чтения. View тела живёт только при отправке полного кадра; до следующего memory.set его scope завершается. На нормальном выходе оба буфера освобождаются логически, stream закрывается, затем закрывается регион. Listener остаётся в main.

Ошибки приложения и ограничения

Коды завершения в --once: 0 — чистый EOF на границе кадров; 2 — неверные аргументы; 3 — I/O, выделение памяти или закрытие; 4 — некорректный кадр. Ошибка закрытия имеет приоритет над протокольной ошибкой. В stderr выводится Invalid frame либо Connection I/O failed. Постоянный сервер после ошибки клиента принимает следующего; ошибка обязательного вывода или accept завершает цикл приёма. Все сообщения диагностики заканчиваются LF.

Один медленный клиент блокирует остальных. Тайм-ауты, readiness, конкурентность, управляемая остановка по сигналу и строгая квота backlog отсутствуют. Следующий этап уже реализован отдельно — неблокирующий reactor. Message Router добавляет адресную маршрутизацию поверх него. Сам blocking-пример сохранён без изменений. Verification Bundle развивается независимо.

Проверки

tests/test_framing.py проверяет библиотеку и приложение при O0/O2:

Все 16384 разбиения 15-байтового потока из трёх сообщений, включая пустое и бинарное; точные длины, смещения, полезные байты и прогресс курсора.
Пустые чанки, приватность и независимость состояния, диапазоны u32, big-endian кодирование и ошибки завершения заголовка/тела.
Реальный loopback: по одному байту, несколько кадров одним потоком, сообщение ровно 65536 байтов, отклонение 65537 и 0xFFFFFFFF.
EOF на каждом префиксе кадра, отсутствие преждевременного ответа, следующий клиент после ошибки предыдущего.
Подмена границы ОС: запись по одному байту, ошибки read/write/close, усечённый заголовок; проверяются точные отправленные байты и однократное закрытие stream и listener.

Набор включён в Linux/macOS compiler tests и Windows acceptance. Это тестовая проверка выбранных сценариев, не формальное доказательство для всех потоков. Парсер, сетевой runtime и машинный код не добавляются к области V0–V4.