Files
cuframes/docs/protocol.md
T
gx ad75aa9624
build / cmake build (CUDA 12.4, Ubuntu 22.04) (pull_request) Successful in 1m35s
build / ffmpeg filter patch (out-of-tree) (pull_request) Successful in 1m39s
docs(protocol): v0.2 — encoded packet ring spec (§10)
Полный wire-protocol spec для encoded packet ring:
- Отдельный SHM /dev/shm/cuframes-<key>-packets (variable-length)
- Backward-compat с v1: proto_version=2 publishers принимают v1 subscribers
- HELLO_REQ/HELLO_RESP extension через reserved bytes — без слома v1 layout
- Codec extradata (SPS/PPS) в shared header
- Late subscriber → keyframe-aligned start (initial_packet_seq)
- Seqlock pattern для защиты от overrun mid-read
- API extension: publish_packet, next_packet, get_codec_params
- 4 новых error codes (OVERSIZED, NO_PACKET_RING, NO_CODEC_PARAMS, PACKET_OVERRUN)

Связано: #2
2026-05-19 16:04:00 +01:00

823 lines
34 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# cuframes Wire Protocol — version 1
**Статус:** v1 — experimental (subject to change до v0.2 release; после — semver-stable)
**Endianness:** little-endian (CUDA-platforms only)
**Дата:** 2026-05-15
Этот документ описывает **byte-exact** формат всех структур и сообщений
cuframes. Любая реализация (C, Python, Rust, FFmpeg filter) должна
соответствовать этому документу.
Связано с: [architecture.md](architecture.md) §2.2.
## 1. Resources / Lifecycle
Один publisher создаёт три kernel-уровневых ресурса:
| Resource | Path | Назначение | Cleanup |
|---|---|---|---|
| Unix socket | `/run/cuframes/<key>.sock` | Handshake + control plane | unlink при destroy() / orphaned после crash — cleanup'ится при next create через `O_EXCL` retry |
| Shared memory | `/dev/shm/cuframes-<key>` | Ring header + slot descriptors | shm_unlink при destroy() / orphaned автоматически если nobody mmap'ит |
| CUDA IPC handles | (no path, через handle in SHM) | Frame data в VRAM | освобождается когда last `cudaIpcCloseMemHandle` (consumer side) |
`<key>` — ASCII, [a-zA-Z0-9_-], 1-63 байт. Library валидирует regex `^[a-zA-Z0-9_-]{1,63}$`.
### Lifecycle при normal shutdown
```
1. cuframes_publisher_destroy() вызван.
2. Publisher шлёт всем connected subscribers: SHUTDOWN message в socket.
3. Subscribers закрывают handles (cudaIpcCloseMemHandle), socket, munmap, return DISCONNECTED.
4. Publisher: cudaIpcClose own pool, close socket, unlink socket+shm.
```
### Lifecycle при abnormal shutdown (publisher crash)
Producer не успевает unlink. Stale socket и shm остаются. Recovery:
```
Next publisher create():
1. Попытка connect(sock_path) → ECONNREFUSED → stale, unlink.
2. Открытие /dev/shm/cuframes-<key>: если magic совпадает, проверка
liveness через pidfd_open(producer_pid) или kill(pid, 0).
3. Если producer dead → shm_unlink + продолжаем create.
4. Если producer жив → return ALREADY_EXISTS.
```
Subscribers detect publisher crash через socket EOF (recv returns 0) или
SIGPIPE на send. После этого они должны cudaIpcCloseMemHandle на всех slot
pointers и return DISCONNECTED в `cuframes_subscriber_next`.
**CRITICAL**: CUDA IPC mem-handles **leak** в consumer'ах если publisher умер
без шанса всем разослать SHUTDOWN. Subscribers ОБЯЗАНЫ Close на DISCONNECTED.
## 2. Shared memory layout
`/dev/shm/cuframes-<key>` имеет фиксированный размер:
`sizeof(cuframes_shared_header_t)`. Без variable-length секций.
### Byte layout (all fields packed, no padding except where shown)
```
Offset Size Field Comments
─────── ────── ────────────────────── ─────────────────────────────────────
0x0000 4 magic (LE u32) 0xCC7C1DCC
0x0004 4 proto_version (LE u32) 1 (см. cuframes_protocol_version())
0x0008 4 lib_version_major
0x000C 4 lib_version_minor
0x0010 4 lib_version_patch
0x0014 4 reserved_a 0
0x0018 8 producer_pid (LE u64) для liveness check
0x0020 8 ring_size (LE u64) 1..16
0x0028 8 ownership_mode 0=LIBRARY, 1=EXTERNAL
0x0030 8 policy 0=DROP_OLDEST, 1=STRICT_WAIT
0x0038 8 max_subscribers 32 (захардкожено в v1)
0x0040 64 meta cuframes_frame_meta packed (см. §2.1)
0x0080 64 ipc_event_handle cudaIpcEventHandle_t (64 байта, opaque NVIDIA blob)
0x00C0 8 global_seq (LE u64) atomic, монотонная
0x00C8 8 subscriber_bitmap atomic, bit per subscriber slot
0x00D0 8 shutdown_flag atomic, 0=normal, 1=publisher shutting down
0x00D8 40 reserved_b 0
0x0100 N×D slots[ring_size] slot descriptor, см. §2.2
0x0100+ND M×S subscribers[32] subscriber slot, см. §2.3
```
Constants:
- N = ring_size, до 16
- D = sizeof(slot_descriptor) = 192 байт (§2.2)
- M = 32 (max subscribers)
- S = sizeof(subscriber_slot) = 128 байт (§2.3)
- Максимальный size SHM: 0x0100 + 16×192 + 32×128 = ~7.3 KB. Headроom для v2 expansion.
Все atomic-помеченные поля доступны через C11 `_Atomic` (или `__atomic_*`).
### 2.1 Frame meta (64 байта)
```
Offset Size Field Comments
0x00 4 format (LE u32) cuframes_format_t enum
0x04 4 width
0x08 4 height
0x0C 4 pitch_y байт на строку Y / single plane
0x10 4 pitch_uv байт на UV (0 если no chroma plane)
0x14 4 bits_per_pixel информативно
0x18 8 frame_size_bytes полный размер (Y+UV или packed)
0x20 32 reserved 0
```
### 2.2 Slot descriptor (192 байта)
```
Offset Size Field Comments
0x00 8 seq (LE u64, atomic) published seq; UINT64_MAX = invalid
0x08 8 pts_ns (LE i64, atomic)
0x10 8 ack_bitmap (LE u64, atomic) bit i = subscriber #i ACK'нул
0x18 8 written_bytes для diagnostics (можно 0)
0x20 64 cuda_mem_handle cudaIpcMemHandle_t (NVIDIA opaque blob)
0x60 32 cuda_ptr_external для OWNERSHIP=EXTERNAL — original
user pointer (informative, not for
consumer use)
0x80 16 reserved_a
0x90 48 reserved_b
0xC0 END
```
Slot — статичный (не меняется после publisher create), за исключением
seq / pts_ns / ack_bitmap / written_bytes которые обновляются атомарно
на каждом publish.
### 2.3 Subscriber slot (128 байт)
В array индексирован bit-position в bitmap (0..31). Создаётся publisher'ом
во время handshake, освобождается на unsubscribe / dead-man timeout.
```
Offset Size Field Comments
0x00 8 state (LE u64, atomic) 0=free, 1=connecting, 2=active, 3=draining
0x08 8 consumer_pid (LE u64) для liveness check
0x10 8 last_seen_seq (LE u64) monotonic — последний seq которое
subscriber успел ACK
0x18 8 last_ack_ns (LE i64) wall-clock последнего ACK
(для consumer_ack_timeout_ms)
0x20 32 consumer_name ASCII zero-terminated, max 31 char
0x40 64 reserved
```
Bit-position **0** зарезервирован — никогда не assigned (sentinel). Используется
bits 1..31 → max 31 subscribers (см. errcode TOO_MANY).
## 3. Unix socket protocol
Publisher `listen()`'ит SOCK_STREAM по `/run/cuframes/<key>.sock`.
Subscriber `connect()`'ится.
Сокет используется для:
- Handshake (subscribe / negotiate)
- Lifetime signals (SHUTDOWN, force-disconnect)
- Late-joining subscribers получают current slot snapshot
Сокет **не** используется для frame transfer — это shared memory + atomic
`global_seq` + (Phase 1+) eventfd для wakeup polling consumers.
### 3.1 Framing
Каждое сообщение — TLV (type-length-value):
```
[4 bytes] message_type (LE u32, see §3.2)
[4 bytes] payload_length (LE u32, bytes excl. these 8 header bytes)
[N bytes] payload (длина = payload_length)
```
`payload_length` max = 4096 для v1 (без длинных сообщений).
### 3.2 Message types
| ID | Name | Direction | Когда |
|---|---|---|---|
| 0x01 | `HELLO_REQ` | C→P | Первое сообщение от consumer |
| 0x02 | `HELLO_RESP` | P→C | Ответ publisher'а |
| 0x03 | `SUBSCRIBE_REQ` | C→P | Завершение handshake, выделить subscriber slot |
| 0x04 | `SUBSCRIBE_RESP` | P→C | Подтверждение + assigned bit |
| 0x10 | `UNSUBSCRIBE` | C→P | Graceful disconnect |
| 0x20 | `EVENT_FD` | P→C | (Phase 1+) FD wake-up channel (sendmsg + cmsg) |
| 0x30 | `SHUTDOWN` | P→C | Publisher shutting down |
| 0xF0 | `PING` | both | Liveness check |
| 0xF1 | `PONG` | both | Reply to PING |
| 0xFE | `ERROR` | both | Error notification (см. §3.3) |
### 3.3 HELLO_REQ payload (consumer → publisher)
```
[4 bytes] proto_version (LE u32) wire-protocol version consumer'а
[4 bytes] consumer_name_len (LE u32)
[N bytes] consumer_name (UTF-8, без null-terminator)
[4 bytes] cuda_device (LE i32)
[4 bytes] mode (LE u32, cuframes_subscriber_mode_t)
[12 bytes] reserved (must be 0)
```
### 3.4 HELLO_RESP payload (publisher → consumer)
```
[4 bytes] result (LE i32) 0 = success, negative = cuframes_error_t
[4 bytes] proto_version_actual (LE u32) — версия publisher'а
[4 bytes] ring_size (LE u32)
[4 bytes] ownership_mode (LE u32)
[64 bytes] frame_meta (см. §2.1)
[4 bytes] shm_path_len (LE u32)
[N bytes] shm_path (UTF-8, ASCII подмножество)
[12 bytes] reserved
```
При result < 0 — payload может быть короче (только result + reserved).
### 3.5 SUBSCRIBE_REQ payload
```
[4 bytes] proto_version (повтор, для double-check)
[12 bytes] reserved
```
### 3.6 SUBSCRIBE_RESP payload
```
[4 bytes] result (LE i32)
[4 bytes] assigned_bit (LE u32) — 1..31
[4 bytes] initial_seq (LE u64) — текущий global_seq на момент handshake;
consumer старт reading с этой точки
[12 bytes] reserved
```
После успешного SUBSCRIBE_RESP — handshake complete. Consumer может poll'ить
`global_seq` в SHM и читать frames.
### 3.7 EVENT_FD (Phase 1+)
Publisher отправляет file descriptor (eventfd) через `sendmsg()` + ancillary
data (`SCM_RIGHTS`). Payload содержит только metadata:
```
[4 bytes] fd_purpose (LE u32) 1 = wake-up на новый seq
[12 bytes] reserved
```
Сам FD приходит в `cmsg`. Consumer `read()`'ит eventfd — блокируется до
producer's `eventfd_write(1)` на следующем publish.
В v1 PoC может быть omitted — consumers poll'ят `global_seq` с usleep.
EVENT_FD добавляется когда poll становится bottleneck'ом (latency).
### 3.8 SHUTDOWN payload
```
[4 bytes] reason (LE u32) 0 = normal, 1 = error, 2 = upgrade
[12 bytes] reserved
```
Subscriber получивший SHUTDOWN → cuda_ipc_close на всех slot pointers,
return DISCONNECTED user'у на next call.
### 3.9 ERROR payload
```
[4 bytes] error_code (LE i32) cuframes_error_t
[4 bytes] message_len (LE u32)
[N bytes] message (UTF-8)
[12 bytes] reserved
```
## 4. State machines
### 4.1 Subscriber-side
```
┌──────────┐ socket connect()
│ STARTUP │────────────────────┐
└──────────┘ ▼
┌─────────┐
│ HELLO │ send HELLO_REQ
└────┬────┘
│ recv HELLO_RESP (ok)
┌─────────┐
│SUBSCRIBE│ send SUBSCRIBE_REQ
└────┬────┘
│ recv SUBSCRIBE_RESP (ok, bit assigned)
┌─────────┐ cuda_ipc_open events / mem
│ READY │
└────┬────┘
socket EOF │ user calls destroy
или SHUTDOWN│ │
▼ ▼
┌─────────────┐
│ CLEANUP │ cuda_ipc_close, unsubscribe send
└─────────────┘
```
### 4.2 Publisher-side per-subscriber
```
recv HELLO_REQ
validate (proto_version, format)
│ │
ok fail
│ └─→ send HELLO_RESP(result=PROTOCOL/FORMAT) → close socket
▼ send HELLO_RESP(ok), wait SUBSCRIBE_REQ
┌─────────────┐
│ HANDSHAKING │
└──────┬──────┘
│ recv SUBSCRIBE_REQ
allocate subscriber_slot, assigned_bit
ok / TOO_MANY?
▼ send SUBSCRIBE_RESP(ok, bit, initial_seq)
┌──────────┐
│ ACTIVE │ — publisher includes этот bit в bitmap requirement
└──────────┘ (для STRICT_WAIT)
subscriber sends UNSUBSCRIBE,
socket EOF, ИЛИ
liveness timeout (PID dead)
release subscriber_slot, clear bit in bitmap
```
## 5. ACK protocol
При публикации slot N publisher:
1. Записывает frame data (или регистрирует external pointer)
2. `cudaEventRecord(event, stream)` на producer stream
3. Atomic: `slot[N].seq = next_seq`, `slot[N].pts_ns = now`, `slot[N].ack_bitmap = 0`
4. Atomic RELEASE: `global_seq = next_seq`
5. (Optional) `eventfd_write(1)` для каждого подписчика
Subscriber:
1. ACQUIRE load `global_seq`. Если новое — process.
2. `cudaStreamWaitEvent(consumer_stream, producer_event, 0)`
3. Process frame (CUDA kernel, copy, etc.)
4. ACK: `atomic_fetch_or(&slot.ack_bitmap, 1 << my_bit, RELEASE)`
5. Atomic store `last_seen_seq = seq`, `last_ack_ns = now`
Publisher (STRICT_WAIT mode) перед next publish ждёт:
- `ack_bitmap == subscriber_bitmap` (все активные ACK'нули), ИЛИ
- timeout `consumer_ack_timeout_ms` истёк → mark dead subscriber, clear bit
Publisher (DROP_OLDEST mode) — не ждёт ACK, просто перезаписывает slot.
## 6. Versioning rules
### 6.1 Wire protocol version
`proto_version` — single integer. v1 — текущий. Breaking changes → bump.
Handshake:
- Если subscriber.proto_version < publisher.proto_version: publisher может
ответить HELLO_RESP с downgraded format (если backward-compat).
v1 — нет backward layer, mismatch → ERROR(PROTOCOL).
- Если subscriber > publisher: publisher отвечает ERROR(PROTOCOL).
### 6.2 Library version (semver)
`lib_version_major.minor.patch` — informational, не используется для
compat decisions. Передаётся в SHM header для diagnostics.
### 6.3 Reserved fields
Все `reserved_*` — должны быть **0** при write. Reader игнорирует.
Это позволяет в minor releases добавлять fields в reserved space без
breaking ABI (т.к. struct sizes остаются те же).
## 7. Conformance test skeleton
Тесты в `tests/conformance/` (Phase 1):
```c
// test_layout.c
TEST(ProtocolLayout, ShmHeaderOffsets) {
EXPECT_EQ(offsetof(cuframes_shared_header_t, magic), 0x0000);
EXPECT_EQ(offsetof(cuframes_shared_header_t, proto_version), 0x0004);
EXPECT_EQ(offsetof(cuframes_shared_header_t, meta), 0x0040);
EXPECT_EQ(offsetof(cuframes_shared_header_t, ipc_event_handle), 0x0080);
// ... see §2 table
}
TEST(ProtocolLayout, SlotDescriptorSize) {
EXPECT_EQ(sizeof(cuframes_slot_descriptor_t), 192);
}
TEST(ProtocolLayout, SubscriberSlotSize) {
EXPECT_EQ(sizeof(cuframes_subscriber_slot_t), 128);
}
```
```c
// test_handshake.c
TEST(Handshake, HelloRespMismatchProto) {
// Setup publisher with proto_version=1
// Connect subscriber with proto_version=2
// Expect: HELLO_RESP(result=ERR_PROTOCOL)
}
```
## 8. Open для v0.2
Эти решения **не** должны нарушить v1 compat:
- TLS / authentication для cross-host scenarios (если когда-нибудь)
- Multi-format в одном publisher (отдельный slot для thumbnail meta)
- AMD/ROCm IPC (заменит cuda_ipc_event_handle на разноименный)
Любое из этих → bump `proto_version` в v2, отдельный document.
## 9. Reference implementation
`libcuframes/src/protocol.c` (Phase 1, Step 2) — единственная reference.
Любая другая реализация (Python ctypes, Rust bindings, FFmpeg plugin)
должна **conformance-tested** против этого документа.
## 10. v0.2 extension: encoded packet ring (proto_version=2)
**Статус:** design draft, ещё не реализовано (см. issue #2).
Параллельно с decoded-frames ring (§2) publisher может опционально
поддерживать **encoded packet ring** — публикует raw H.264/H.265 NAL units
**до** decoder, для consumer'ов которые делают `-c:v copy` (recording, mux).
### 10.1 Совместимость с v1
- v2 publisher принимает **v1-subscribers** — они получают только frames
ring (как v0.1), packet ring им не показывается.
- v1 publisher отвергает v2-subscribers с `wants_packets=true`
(HELLO_RESP error PROTOCOL).
- v1 layout (§2) **не меняется** для frames ring — packet ring это отдельный SHM.
Publisher version bumping:
- `proto_version` = 2 в SHM header и в HELLO_RESP когда packet ring active.
- Если publisher v2 не активирует packet ring (`enable_packet_ring=false`)
`proto_version` остаётся 1 (полная v1 compat).
### 10.2 Дополнительные ресурсы
| Resource | Path | Назначение | Когда |
|---|---|---|---|
| Packet shared memory | `/dev/shm/cuframes-<key>-packets` | Packet ring header + slots + byte buffer | если publisher активировал packet ring |
Cleanup — симметрично §1: `shm_unlink` при destroy(); orphaned автоматически
если nobody mmap'ит.
### 10.3 Packet ring layout
Размер пакетного SHM: `sizeof(packet_ring_header_t) + N×PSE + DATA_SIZE`,
где:
- N = `packet_ring_slots`, default 64 (configurable)
- PSE = `sizeof(packet_slot_entry_t)` = 64 байт (см. §10.5)
- DATA_SIZE = `packet_data_size`, default 8 MB (configurable)
#### Byte layout
```
Offset Size Field Comments
─────────────────────── ────── ────────────────────────── ─────────────────────────────
0x0000 4 magic (LE u32) 0xCC7C1DCD (frames magic + 1)
0x0004 4 proto_version (LE u32) 2
0x0008 4 ring_slots (LE u32) N (1..1024)
0x000C 4 data_size (LE u32) bytes for packet data ring
0x0010 4 codec_id (LE u32) AV_CODEC_ID_* enum (см. §10.4)
0x0014 4 codec_extradata_size (LE u32) ≤ 4096
0x0018 8 producer_pid (LE u64)
0x0020 8 global_seq (LE u64, atomic) монотонная по packets
0x0028 8 last_keyframe_seq (LE u64, atomic) для late subscribers
0x0030 8 write_offset (LE u64, atomic) текущий cursor в data ring
0x0038 8 shutdown_flag (LE u64, atomic)
0x0040 4096 codec_extradata SPS/PPS/VPS bytes (см. §10.4)
0x1040 N×64 slots[N] packet_slot_entry_t (см. §10.5)
0x1040+N×64 DATA_SIZE data[] wraparound byte buffer
```
Все atomic fields — C11 `_Atomic` (release/acquire semantics для seq updates).
### 10.4 Codec extradata
H.264 — SPS + PPS, конкатенированные в **Annex B** формате
(start codes `00 00 00 01`). H.265 — VPS + SPS + PPS.
`codec_id` соответствует FFmpeg `AV_CODEC_ID_H264`, `AV_CODEC_ID_HEVC`,
`AV_CODEC_ID_AV1` (future). Subscriber пишет этот extradata в
`AVCodecContext.extradata` своего decoder'а (если он его создаёт)
или в `AVStream.codecpar->extradata` для muxer'ов.
Extradata устанавливается publisher'ом **один раз** при первом keyframe
(или из RTSP SDP до первого packet). После — fixed на lifetime publisher'а
(codec change mid-stream → publisher destroy+recreate с новым `<key>`).
### 10.5 Packet slot entry (64 байта)
```
Offset Size Field Comments
0x00 8 seq (LE u64, atomic) published seq; UINT64_MAX = invalid
0x08 8 pts_ns (LE i64)
0x10 8 dts_ns (LE i64) для B-frames pipelines
0x18 8 data_offset (LE u64) offset в `data[]` секции SHM
0x20 4 data_size (LE u32) size of payload bytes
0x24 4 flags (LE u32) §10.6
0x28 24 reserved 0
```
`data_offset` может быть **больше** `data_size` секции SHM — semantics
"absolute byte cursor", фактический byte index = `data_offset % data_size`.
Subscriber может detect wrap (если payload crosses end → split read).
### 10.6 Packet flags
```
Bit Name Comments
0 KEY keyframe (IDR for H.264, или CRA/IDR для HEVC).
Critical для late subscribers — must wait IDR.
1 CORRUPT publisher detect'нул что packet damaged
(RTP loss и т.п.) — subscriber может skip
2 DISCONTINUITY был gap перед этим packet
(publisher reconnect к камере)
3 LAST_IN_AU last NAL в access unit (полный frame)
— для muxer'ов которые ждут полный frame
4-31 reserved 0
```
Mapping в `AVPacket.flags`:
- bit 0 (KEY) → `AV_PKT_FLAG_KEY`
- bit 1 (CORRUPT) → `AV_PKT_FLAG_CORRUPT`
- bit 2 (DISCONTINUITY) → `AV_PKT_FLAG_DISCONTINUITY` (FFmpeg 5+)
### 10.7 Atomic publish (publisher-side)
```c
// Pseudo-C (упрощено, без error handling)
uint64_t seq = atomic_load(&hdr->global_seq, RELAXED) + 1;
uint64_t off = atomic_load(&hdr->write_offset, RELAXED);
// 1. Найти free slot (overwrite oldest)
size_t slot_idx = seq % hdr->ring_slots;
packet_slot_entry_t *slot = &slots[slot_idx];
// 2. Записать payload bytes (wraparound, может потребовать 2 memcpy)
size_t off_in_ring = off % hdr->data_size;
size_t first_chunk = min(size, hdr->data_size - off_in_ring);
memcpy(data + off_in_ring, payload, first_chunk);
if (first_chunk < size)
memcpy(data, payload + first_chunk, size - first_chunk);
// 3. RELEASE: записать metadata в slot
slot->pts_ns = pts;
slot->dts_ns = dts;
slot->data_offset = off;
slot->data_size = size;
slot->flags = flags;
atomic_store(&slot->seq, seq, RELEASE);
// 4. Update global cursor + global_seq
atomic_store(&hdr->write_offset, off + size, RELEASE);
atomic_store(&hdr->global_seq, seq, RELEASE);
// 5. If KEY → update last_keyframe_seq
if (flags & PKT_FLAG_KEY)
atomic_store(&hdr->last_keyframe_seq, seq, RELEASE);
```
### 10.8 Atomic read (subscriber-side)
```c
// Pseudo-C
uint64_t cur = atomic_load(&hdr->global_seq, ACQUIRE);
if (cur <= my_last_seq) return TIMEOUT; // ничего нового
uint64_t want_seq = my_last_seq + 1;
size_t slot_idx = want_seq % hdr->ring_slots;
packet_slot_entry_t *slot = &slots[slot_idx];
uint64_t slot_seq = atomic_load(&slot->seq, ACQUIRE);
if (slot_seq != want_seq) {
// overrun — slow subscriber. Re-anchor:
want_seq = atomic_load(&hdr->last_keyframe_seq, ACQUIRE);
slot_idx = want_seq % hdr->ring_slots;
slot = &slots[slot_idx];
return DROPPED; // signal user через flags = DISCONTINUITY
}
// Copy payload (wraparound aware)
uint64_t off = slot->data_offset % hdr->data_size;
uint32_t size = slot->data_size;
uint32_t first_chunk = min(size, hdr->data_size - off);
memcpy(out_buf, data + off, first_chunk);
if (first_chunk < size)
memcpy(out_buf + first_chunk, data, size - first_chunk);
// Re-check slot->seq не изменился (защита от overrun mid-read)
if (atomic_load(&slot->seq, ACQUIRE) != want_seq) {
return DROPPED; // publisher overwrote во время copy
}
my_last_seq = want_seq;
return OK;
```
Защита от overrun mid-read через **post-check `slot->seq`** — простая
вариант seqlock. Если publisher успел overwrite между metadata-read и
data-copy — subscriber detect и retry.
### 10.9 Socket protocol extensions
#### HELLO_REQ — добавляются flags в reserved field
v1 layout (§3.3):
```
[4 bytes] proto_version
[4 bytes] consumer_name_len
[N bytes] consumer_name
[4 bytes] cuda_device
[4 bytes] mode
[12 bytes] reserved (must be 0) ← v0.2 использует первые 4 байта
```
v0.2 интерпретирует первые 4 байта `reserved` как `subscribe_flags`:
| Bit | Name | Comments |
|---|---|---|
| 0 | `WANTS_FRAMES` | подписаться на decoded frames ring (default ON в v1 — implicit) |
| 1 | `WANTS_PACKETS` | подписаться на encoded packet ring |
| 2-31 | reserved | 0 |
Если v1-subscriber оставляет reserved=0 — publisher v2 интерпретирует это
как `WANTS_FRAMES=true, WANTS_PACKETS=false` (v1 backward-compat).
#### HELLO_RESP — добавляются packet-ring fields
v1 layout (§3.4) расширяется в reserved секции:
```
[4 bytes] result
[4 bytes] proto_version_actual ← теперь может быть 1 или 2
[4 bytes] ring_size ← frames ring
[4 bytes] ownership_mode
[64 bytes] frame_meta
[4 bytes] shm_path_len ← frames SHM
[N bytes] shm_path
[12 bytes] reserved ← v0.2 интерпретирует
```
v0.2 reserved layout (если `proto_version_actual == 2` И publisher
поддерживает packets):
```
[4 bytes] packet_shm_path_len (LE u32) 0 = packets disabled at publisher
[N bytes] packet_shm_path (UTF-8) — относительно /dev/shm/, например "cuframes-camA-packets"
[4 bytes] codec_id (LE u32) AV_CODEC_ID_*
[4 bytes] initial_packet_seq (LE u64) last_keyframe_seq на момент handshake
(subscriber должен start с этого seq)
```
Если subscriber запросил `WANTS_PACKETS=1` но publisher не имеет packet ring
`result = ERR_NOT_AVAILABLE`.
### 10.10 Subscriber state machine extension
Подключение к **обоим** rings (или одному из):
```
┌──────────┐
│ HELLO_OK │ proto_version_actual=2, packet_shm_path_len>0
└────┬─────┘
┌────────────────────────────────┐
│ Open frames SHM (если WANTS_FRAMES) │ → standard v1 flow
└────────────────────────────────┘
┌────────────────────────────────┐
│ Open packet SHM (если WANTS_PACKETS) │
│ - mmap /dev/shm/cuframes-<key>-packets │
│ - check magic, proto_version │
│ - set my_last_packet_seq = initial_packet_seq - 1 │
│ (так что первый next_packet вернёт IDR) │
└────────────────────────────────┘
┌─────────┐
│ READY │ — frames или packets или оба доступны
└─────────┘
```
### 10.11 Threading в subscriber
Frames ring и packet ring имеют **разные** `global_seq` counters.
Subscriber имеет **отдельные** `my_last_seq` для каждого. Может
poll'ить обе независимо (или через два threads).
Producer's `cudaEventRecord` (frames sync) не релевантен для packets —
encoded data на CPU, без CUDA sync.
### 10.12 Конфигурируемость packet ring
Publisher API extension (§10.13) принимает параметры:
```c
typedef struct {
uint32_t packet_ring_slots; // default 64
uint32_t packet_data_size; // default 8 MB (8388608)
uint32_t max_packet_size; // default 2 MB — sanity guard для оversized
// packets (publisher rejects with error)
uint32_t codec_id; // AV_CODEC_ID_H264 / HEVC / ...
} cuframes_packet_ring_options_t;
```
### 10.13 API extension (для cuframes.h)
```c
/* Сreate publisher с активным packet ring. NULL для opts → packet ring disabled. */
int cuframes_publisher_create_ex(
const cuframes_publisher_options_t *frames_opts,
const cuframes_packet_ring_options_t *packet_opts, /* NULL = no packet ring */
cuframes_publisher_t **pub_out
);
/* Set codec extradata (SPS/PPS) — должен быть called до первого publish_packet. */
int cuframes_publisher_set_codec_extradata(
cuframes_publisher_t *pub,
const void *extradata,
size_t size
);
/* Публикация packet. Slow consumer = overwrite oldest. */
int cuframes_publisher_publish_packet(
cuframes_publisher_t *pub,
const void *data,
size_t size,
int64_t pts_ns,
int64_t dts_ns,
uint32_t flags /* CUFRAMES_PKT_FLAG_KEY | _CORRUPT | _DISCONTINUITY | _LAST_IN_AU */
);
/* Subscriber-side: подписаться с opt-in для packets. */
typedef struct {
/* ... existing v1 fields ... */
uint32_t subscribe_flags; /* WANTS_FRAMES, WANTS_PACKETS bits */
} cuframes_subscriber_options_v2_t;
int cuframes_subscriber_create_v2(
const cuframes_subscriber_options_v2_t *opts,
cuframes_subscriber_t **sub_out
);
/* Чтение packet. Opaque handle — каллер вызывает release_packet после. */
typedef struct cuframes_packet cuframes_packet_t;
int cuframes_subscriber_next_packet(
cuframes_subscriber_t *sub,
cuframes_packet_t **pkt_out,
int32_t timeout_ms
);
const void * cuframes_packet_data(const cuframes_packet_t *p);
size_t cuframes_packet_size(const cuframes_packet_t *p);
int64_t cuframes_packet_pts(const cuframes_packet_t *p);
int64_t cuframes_packet_dts(const cuframes_packet_t *p);
uint32_t cuframes_packet_flags(const cuframes_packet_t *p);
int cuframes_subscriber_release_packet(cuframes_subscriber_t *sub, cuframes_packet_t *p);
/* Codec params для subscriber (extracted из shared header). */
int cuframes_subscriber_get_codec_params(
cuframes_subscriber_t *sub,
uint32_t *codec_id_out,
const void **extradata_out,
size_t *extradata_size_out
);
```
`cuframes_packet_t` opaque — фактически указатель в local-mapped data (на
heap subscriber'а — copy при `next_packet`, освобождение при `release`).
Subscriber **не** держит ссылки на shared ring data между `next_packet` и
`release_packet` — это избавляет от reader-locks.
### 10.14 Late subscriber → keyframe-aligned start
При SUBSCRIBE_RESP publisher отвечает `initial_packet_seq = last_keyframe_seq`.
Subscriber устанавливает `my_last_seq = initial_packet_seq - 1`, так что
первый `next_packet` вернёт keyframe (decoder может start без glitches).
**Risk:** если в момент handshake **last_keyframe_seq уже выехал из
ring** (slow start subscriber, GOP > ring_slots packets) — subscriber
detect overrun в первом read и переходит на следующий keyframe.
В implementation `publisher_publish_packet` для оптимизации может маркировать
slot перед IDR как **persistent** (флаг в reserved), но **v0.2 keep simple**
просто требуем что `packet_ring_slots × avg_packet_size > GOP_size_in_bytes`
для нормальной работы. Sizing guide см. в [docs/integration.md](integration.md).
### 10.15 Error codes (новые)
| Code | Name | Когда |
|---|---|---|
| -20 | `CUFRAMES_ERR_PACKET_OVERSIZED` | publish_packet с size > max_packet_size |
| -21 | `CUFRAMES_ERR_NO_PACKET_RING` | subscriber запросил packets, publisher без packet ring |
| -22 | `CUFRAMES_ERR_NO_CODEC_PARAMS` | get_codec_params вызван до set_codec_extradata publisher'ом |
| -23 | `CUFRAMES_ERR_PACKET_OVERRUN` | subscriber slow — packet seq уехал, надо resync на keyframe |
### 10.16 Open для v0.3+
- **Sub-stream selection** — publisher может публиковать несколько
packet rings (для multi-resolution streams). Сейчас один key = один stream.
v0.3 → `<key>-substream-<N>` naming?
- **Codec change mid-stream** — текущий design требует publisher restart.
Future: invalidate codec_extradata + bump generation field.
- **Audio streams** — analogichno в packet ring, но codec_id = audio (AAC,
Opus). v0.3.