cuframes-docs/site/i18n/ru/docusaurus-plugin-content-docs/current/getting-started/first-subscriber.md

title, sidebar_position

title	sidebar_position
Первый subscriber	3

Первый subscriber

Минимальный subscriber, который подключается к publisher'у из Первого publisher, читает 10 frame'ов и проверяет, что каждый байт каждого frame'а совпадает с pattern'ом со стороны publisher'а.

Это упрощённая версия spike/smoke_v04/smoke_sub.c.

Исходник

/* first_subscriber.c — connect, read 10 frames, verify pattern. */
#include <cuframes/cuframes.h>
#include <cuda_runtime.h>
#include <stdio.h>
#include <stdlib.h>

int main(int argc, char **argv) {
    const char *key = argc > 1 ? argv[1] : "mykey";

    cuframes_subscriber_config_t cfg = {0};
    cfg.key                = key;
    cfg.consumer_name      = "first-sub";
    cfg.mode               = CUFRAMES_MODE_NEWEST_ONLY;
    cfg.cuda_device        = 0;
    cfg.connect_timeout_ms = 5000;

    cuframes_subscriber_t *sub = NULL;
    int r = cuframes_subscriber_create(&cfg, &sub);
    if (r != CUFRAMES_OK) {
        fprintf(stderr, "create: %s\n", cuframes_strerror(r));
        return 1;
    }

    cudaStream_t stream;
    cudaStreamCreate(&stream);

    const size_t sample = 1024;             /* check first 1 KiB of each frame */
    uint8_t *host = malloc(sample);

    int frames = 0, good = 0;
    while (frames < 10) {
        cuframes_frame_t *f = NULL;
        r = cuframes_subscriber_next(sub, stream, &f, 2000);
        if (r != CUFRAMES_OK) {
            fprintf(stderr, "next: %s\n", cuframes_strerror(r));
            break;
        }

        cudaMemcpyAsync(host, cuframes_frame_cuda_ptr(f), sample,
                        cudaMemcpyDeviceToHost, stream);
        cudaStreamSynchronize(stream);

        int mismatch = 0;
        for (size_t i = 1; i < sample; i++)
            if (host[i] != host[0]) mismatch++;
        if (mismatch == 0) good++;

        printf("seq=%lu pts_ns=%lld pitch_y=%d byte0=0x%02x mismatch=%d\n",
               (unsigned long)cuframes_frame_seq(f),
               (long long)cuframes_frame_pts_ns(f),
               cuframes_frame_pitch_y(f),
               host[0], mismatch);

        cuframes_subscriber_release(sub, f);
        frames++;
    }

    free(host);
    cudaStreamDestroy(stream);
    cuframes_subscriber_destroy(sub);
    return (good == frames && frames > 0) ? 0 : 1;
}

Разбор

cfg.key — должен точно совпадать с key publisher'а. Subscriber находит publisher'а через connect() на /run/cuframes/<key>.sock.

cfg.consumer_name = "first-sub" — идентифицирует subscriber'а в ACK-bitmap publisher'а. Должен быть уникален среди живых subscriber'ов одного publisher'а; коллизия возвращает CUFRAMES_ERR_ALREADY_EXISTS. Если передать NULL, library сгенерирует subscriber-<pid>-<random>. Publisher принимает до 32 одновременных subscriber'ов.

cfg.mode = CUFRAMES_MODE_NEWEST_ONLY — subscriber всегда прыгает к самому свежему опубликованному frame'у и пропускает все frame'ы, которые publisher успел произвести пока обрабатывался предыдущий вызов next(). Используй CUFRAMES_MODE_STRICT_ORDER, если обязательно нужно видеть каждый frame по seq; в этом режиме ring overflow всплывает как CUFRAMES_ERR_DISCONNECTED.

cfg.connect_timeout_ms = 5000 — сколько create() ждёт появления publisher'а. 0 — fail immediately с CUFRAMES_ERR_NOT_FOUND, -1 — ждать вечно.

cfg.cuda_device — должен совпадать с cuda_device publisher'а. CUDA IPC handle'ы не переносимы между девайсами.

cuframes_subscriber_next(sub, stream, &f, 2000) — блокируется до 2 секунд ради следующего frame'а. Library внутри вызывает cudaStreamWaitEvent на твоём stream'е против publisher'овского record-event, поэтому любой kernel, запущенный на stream'е после возврата next(), гарантированно увидит writes producer'а. Если читаешь через cudaMemcpyDeviceToHost, ставь его в очередь на тот же stream — это и делает cross-process sync рабочим.

Frame-accessor'ы — cuframes_frame_cuda_ptr() (device pointer, read-only), cuframes_frame_format(), cuframes_frame_pitch_y() / _pitch_uv(), cuframes_frame_seq() (монотонный per publisher), cuframes_frame_pts_ns() (CLOCK_MONOTONIC со стороны publisher'а). После release() handle становится недействительным — не вызывай на нём accessor'ы.

cuframes_subscriber_release(sub, f) — отдаёт ACK на slot publisher'у. Publisher'у это нужно только при CUFRAMES_POLICY_STRICT_WAIT; при дефолтном DROP_OLDEST всё равно обязательно вызывать, чтобы освободить handle на стороне consumer'а. NULL — no-op.

Компиляция

gcc -O2 -I/usr/local/include -I/usr/local/cuda/include \
    -o first_subscriber first_subscriber.c \
    -L/usr/local/lib -lcuframes \
    -L/usr/local/cuda/lib64 -lcudart -lcuda

Запуск

В одном терминале запусти publisher с предыдущей страницы:

./first_publisher mykey

В другом терминале:

./first_subscriber mykey

Ожидаемый вывод — 10 строк с растущим seq и mismatch=0.

Замечание про Docker

Subscriber должен шарить IPC namespace publisher'а (чтобы shm_open смог открыть тот же header /dev/shm/cuframes-mykey). PID namespace шарить не нужно — это изменение v0.4. Старое требование v0.1 / v0.2 --pid=container:<publisher> ушло, потому что handle'ы ездят как POSIX file descriptors через Unix socket (SCM_RIGHTS), а не как CUDA IPC mem-handle'ы.

docker run --rm --runtime=nvidia \
  --ipc=container:cuframes-pub \
  -v /run/cuframes:/run/cuframes:ro \
  gx/cuframes:0.4 ./first_subscriber mykey

Обработка disconnect

Если publisher завершился или упал во время твоего loop'а, следующий cuframes_subscriber_next() вернёт CUFRAMES_ERR_DISCONNECTED. Handle после этого мёртв — уничтожь его и (опционально) переподключись:

if (r == CUFRAMES_ERR_DISCONNECTED) {
    cuframes_subscriber_destroy(sub);
    sub = NULL;
    /* sleep + retry cuframes_subscriber_create(&cfg, &sub) */
}

Паттерн reconnect, включая back-off и нюансы переиспользования consumer_name, : поймать CUFRAMES_ERR_DISCONNECTED, вызвать cuframes_subscriber_destroy(), подождать (1-2 сек) и попробовать cuframes_subscriber_create() снова с тем же key. FFmpeg-демухер cuframes:// делает это автоматически (см. Интеграция → FFmpeg demuxer).