/*
* Copyright (C) 2023 Mikhail Burakov. This file is part of receiver.
*
* receiver is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* receiver is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with receiver. If not, see .
*/
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include "audio.h"
#include "decode.h"
#include "input.h"
#include "proto.h"
#include "pui/font.h"
#include "toolbox/buffer.h"
#include "toolbox/perf.h"
#include "toolbox/utils.h"
#include "window.h"
static volatile sig_atomic_t g_signal;
static void OnSignal(int status) { g_signal = status; }
struct Context {
struct InputStream* input_stream;
struct Window* window;
size_t overlay_width;
size_t overlay_height;
struct Overlay* overlay;
struct DecodeContext* decode_context;
struct AudioContext* audio_context;
struct Buffer buffer;
size_t video_bitstream;
size_t audio_bitstream;
uint64_t timestamp;
uint64_t ping_sum;
uint64_t ping_count;
uint64_t video_latency_sum;
uint64_t video_latency_count;
uint64_t audio_latency_sum;
uint64_t audio_latency_count;
};
static int ConnectSocket(const char* arg) {
uint16_t port;
char ip[sizeof("xxx.xxx.xxx.xxx")];
if (sscanf(arg, "%[0-9.]:%hu", ip, &port) != 2) {
LOG("Failed to parse address");
return -1;
}
int sock = socket(AF_INET, SOCK_STREAM, 0);
if (sock == -1) {
LOG("Failed to create socket (%s)", strerror(errno));
return -1;
}
if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, &(int){1}, sizeof(int))) {
LOG("Failed to set TCP_NODELAY (%s)", strerror(errno));
goto rollback_sock;
}
// TODO(mburakov): Set and maintain TCP_QUICKACK.
const struct sockaddr_in addr = {
.sin_family = AF_INET,
.sin_port = htons(port),
.sin_addr.s_addr = inet_addr(ip),
};
if (connect(sock, (const struct sockaddr*)&addr, sizeof(addr))) {
LOG("Failed to connect socket (%s)", strerror(errno));
goto rollback_sock;
}
return sock;
rollback_sock:
close(sock);
return -1;
}
static void OnWindowClose(void* user) {
(void)user;
g_signal = SIGINT;
}
static void OnWindowFocus(void* user, bool focused) {
if (focused) return;
if (!InputStreamHandsoff(user)) {
LOG("Failed to handle window focus");
g_signal = SIGABRT;
}
}
static void OnWindowKey(void* user, unsigned key, bool pressed) {
if (!InputStreamKeyPress(user, key, pressed)) {
LOG("Failed to handle key press");
g_signal = SIGABRT;
}
}
static void OnWindowMove(void* user, int dx, int dy) {
if (!InputStreamMouseMove(user, dx, dy)) {
LOG("Failed to handle mouse move");
g_signal = SIGABRT;
}
}
static void OnWindowButton(void* user, unsigned button, bool pressed) {
if (!InputStreamMouseButton(user, button, pressed)) {
LOG("Failed to handle mouse button");
g_signal = SIGABRT;
}
}
static void OnWindowWheel(void* user, int delta) {
if (!InputStreamMouseWheel(user, delta)) {
LOG("Failed to handle mouse wheel");
g_signal = SIGABRT;
}
}
static void GetMaxOverlaySize(size_t* width, size_t* height) {
char str[64];
snprintf(str, sizeof(str), "Video bitstream: %zu.000 Mbps", SIZE_MAX / 1000);
*width = 4 + PuiStringWidth(str) + 4;
*height = 4 + 12 * 5 + 4;
}
static struct Context* ContextCreate(int sock, bool no_input, bool stats,
const char* audio_device) {
struct Context* context = calloc(1, sizeof(struct Context));
if (!context) {
LOG("Failed to allocate context (%s)", strerror(errno));
return NULL;
}
const struct WindowEventHandlers* maybe_window_event_handlers = NULL;
if (!no_input) {
context->input_stream = InputStreamCreate(sock);
if (!context->input_stream) {
LOG("Failed to create input stream");
goto rollback_context;
}
static const struct WindowEventHandlers window_event_handlers = {
.OnClose = OnWindowClose,
.OnFocus = OnWindowFocus,
.OnKey = OnWindowKey,
.OnMove = OnWindowMove,
.OnButton = OnWindowButton,
.OnWheel = OnWindowWheel,
};
maybe_window_event_handlers = &window_event_handlers;
}
context->window =
WindowCreate(maybe_window_event_handlers, context->input_stream);
if (!context->window) {
LOG("Failed to create window");
goto rollback_input_stream;
}
if (stats) {
GetMaxOverlaySize(&context->overlay_width, &context->overlay_height);
context->overlay =
OverlayCreate(context->window, 4, 4, (int)context->overlay_width,
(int)context->overlay_height);
if (!context->overlay) {
LOG("Failed to create stats overlay");
goto rollback_window;
}
}
context->decode_context = DecodeContextCreate(context->window);
if (!context->decode_context) {
LOG("Failed to create decode context");
goto rollback_overlay;
}
if (audio_device) {
context->audio_context = AudioContextCreate(audio_device);
if (!context->audio_context) {
LOG("Failed to create audio context");
goto rollback_decode_context;
}
}
return context;
rollback_decode_context:
DecodeContextDestroy(context->decode_context);
rollback_overlay:
if (context->overlay) OverlayDestroy(context->overlay);
rollback_window:
WindowDestroy(context->window);
rollback_input_stream:
if (context->input_stream) InputStreamDestroy(context->input_stream);
rollback_context:
free(context);
return NULL;
}
static bool RenderOverlay(struct Context* context, uint64_t timestamp) {
uint32_t* buffer = OverlayLock(context->overlay);
if (!buffer) {
LOG("Failed to lock overlay");
return false;
}
char ping_str[64];
uint64_t ping = 0;
if (context->ping_count) {
ping = context->ping_sum / context->ping_count;
}
snprintf(ping_str, sizeof(ping_str), "Ping: %zu.%03zu ms", ping / 1000,
ping % 1000);
char video_bitrate_str[64];
uint64_t clock_delta = timestamp - context->timestamp;
// mburakov: Kbps = nbytes * 1sec * 8bit / clock_delta / 1024
size_t video_bitrate =
context->video_bitstream * 1000000 * 8 / clock_delta / 1024;
snprintf(video_bitrate_str, sizeof(video_bitrate_str),
"Video bitrate: %zu.%03zu Mbps", video_bitrate / 1000,
video_bitrate % 1000);
char audio_bitrate_str[64];
size_t audio_bitrate = 0;
if (context->audio_context) {
// mburakov: Kbps = nbytes * 1sec * 8bit / clock_delta / 1024
audio_bitrate = context->audio_bitstream * 1000000 * 8 / clock_delta / 1024;
snprintf(audio_bitrate_str, sizeof(audio_bitrate_str),
"Audio bitrate: %zu.%03zu Mbps", audio_bitrate / 1000,
audio_bitrate % 1000);
}
char video_latency_str[64];
uint64_t video_latency = 0;
if (context->video_latency_count) {
// mburakov: Pessimistic calculations, these assume one fully missed vsync
// for capture, one fully missed vsync for rendering, and 100Mbit network.
// latency = avg_latency + ping + vsync + vsync + Kbps * 1sec / 100Mbps
video_latency =
context->video_latency_sum / context->video_latency_count + ping +
16666 + 16666 +
video_bitrate * 1000000 / 100000000 / context->video_latency_count;
}
snprintf(video_latency_str, sizeof(video_latency_str),
"Video latency: %zu.%03zu ms", video_latency / 1000,
video_latency % 1000);
char audio_latency_str[64];
if (context->audio_context) {
uint64_t audio_latency = 0;
if (context->audio_latency_count) {
// mburakov: Pessimistic calculations, assume 100Mbit network. Capture
// and playback periods are unknown, but should roughly correspond to the
// latency reported by the audio context, because it commulatively
// includes all the missed periods since the beginning of streming.
// latency = avg_latency + ping + Kbps * 1sec / 100Mbps + context_latency
audio_latency =
context->audio_latency_sum / context->audio_latency_count + ping +
audio_bitrate * 1000000 / 100000000 +
AudioContextGetLatency(context->audio_context);
}
snprintf(audio_latency_str, sizeof(audio_latency_str),
"Audio latency: %zu.%03zu ms", audio_latency / 1000,
audio_latency % 1000);
}
char* lines[5] = {NULL};
char** plines = lines;
*plines++ = ping_str;
*plines++ = video_bitrate_str;
if (context->audio_context) *plines++ = audio_bitrate_str;
*plines++ = video_latency_str;
if (context->audio_context) *plines++ = audio_latency_str;
size_t nlines = (size_t)(plines - lines);
size_t overlay_width = 0;
for (size_t i = 0; i < nlines; i++)
overlay_width = MAX(overlay_width, PuiStringWidth(lines[i]));
overlay_width += 8;
size_t overlay_height = 12 * nlines + 8;
memset(buffer, 0, context->overlay_width * context->overlay_height * 4);
for (size_t y = 0; y < overlay_height; y++) {
for (size_t x = 0; x < overlay_width; x++)
buffer[x + y * context->overlay_width] = 0x40000000;
}
for (size_t i = 0; i < nlines; i++) {
size_t voffset = context->overlay_width * (4 + 12 * i);
PuiStringRender(lines[i], buffer + voffset + 4, context->overlay_width,
0xffffffff);
}
OverlayUnlock(context->overlay);
return true;
}
static bool HandleVideoStream(struct Context* context) {
const struct Proto* proto = context->buffer.data;
if (!DecodeContextDecode(context->decode_context, proto->data, proto->size)) {
LOG("Failed to decode incoming video data");
return false;
}
if (!context->overlay) return true;
if (!context->timestamp) {
context->timestamp = MicrosNow();
return true;
}
context->video_bitstream += proto->size;
context->video_latency_sum += proto->latency;
context->video_latency_count++;
if (!(proto->flags & PROTO_FLAG_KEYFRAME)) return true;
uint64_t timestamp = MicrosNow();
if (!RenderOverlay(context, timestamp)) LOG("Failed to render overlay");
context->video_bitstream = 0;
context->audio_bitstream = 0;
context->timestamp = timestamp;
context->ping_sum = 0;
context->ping_count = 0;
context->video_latency_sum = 0;
context->video_latency_count = 0;
context->audio_latency_sum = 0;
context->audio_latency_count = 0;
return true;
}
static bool HandleAudioStream(struct Context* context) {
const struct Proto* proto = context->buffer.data;
if (!context->audio_context) return true;
if (!AudioContextDecode(context->audio_context, proto->data, proto->size)) {
LOG("Failed to decode incoming audio data");
return false;
}
if (!context->overlay) return true;
if (!context->timestamp) {
context->timestamp = MicrosNow();
return true;
}
context->audio_bitstream += proto->size;
context->audio_latency_sum += proto->latency;
context->audio_latency_count++;
return true;
}
static bool DemuxProtoStream(int sock, struct Context* context) {
switch (BufferAppendFrom(&context->buffer, sock)) {
case -1:
LOG("Failed to append packet data to buffer (%s)", strerror(errno));
return false;
case 0:
LOG("Server closed connection");
return false;
default:
break;
}
again:
if (context->buffer.size < sizeof(struct Proto)) return true;
const struct Proto* proto = context->buffer.data;
if (context->buffer.size < sizeof(struct Proto) + proto->size) return true;
switch (proto->type) {
case PROTO_TYPE_MISC:
context->ping_sum +=
MicrosNow() - *(const uint64_t*)(const void*)proto->data;
context->ping_count++;
break;
case PROTO_TYPE_VIDEO:
if (!HandleVideoStream(context)) {
LOG("Failed to handle video stream");
return false;
}
break;
case PROTO_TYPE_AUDIO:
if (!HandleAudioStream(context)) {
LOG("Failed to handle audio stream");
return false;
}
}
BufferDiscard(&context->buffer, sizeof(struct Proto) + proto->size);
goto again;
}
static bool SendPingMessage(int sock, int timer_fd, struct Context* context) {
uint64_t expirations;
if (read(timer_fd, &expirations, sizeof(expirations)) !=
sizeof(expirations)) {
LOG("Failed to read timer expirations (%s)", strerror(errno));
return false;
}
struct {
uint32_t type;
uint64_t timestamp;
} __attribute__((packed)) ping = {
.type = ~0u,
.timestamp = MicrosNow(),
};
if (write(sock, &ping, sizeof(ping)) != sizeof(ping)) {
LOG("Failed to write ping message (%s)", strerror(errno));
return false;
}
return true;
}
static void ContextDestroy(struct Context* context) {
BufferDestroy(&context->buffer);
if (context->audio_context) AudioContextDestroy(context->audio_context);
DecodeContextDestroy(context->decode_context);
if (context->overlay) OverlayDestroy(context->overlay);
WindowDestroy(context->window);
if (context->input_stream) InputStreamDestroy(context->input_stream);
}
int main(int argc, char* argv[]) {
if (argc < 2) {
LOG("Usage: %s : [--no-input] [--stats] [--audio ]",
argv[0]);
return EXIT_FAILURE;
}
int sock = ConnectSocket(argv[1]);
if (sock == -1) {
LOG("Failed to connect socket");
return EXIT_FAILURE;
}
bool no_input = false;
bool stats = false;
const char* audio_device = NULL;
for (int i = 2; i < argc; i++) {
if (!strcmp(argv[i], "--no-input")) {
no_input = true;
} else if (!strcmp(argv[i], "--stats")) {
stats = true;
} else if (!strcmp(argv[i], "--audio")) {
audio_device = argv[++i];
if (i == argc) {
LOG("Audio argument requires a value");
return EXIT_FAILURE;
}
}
}
struct Context* context = ContextCreate(sock, no_input, stats, audio_device);
if (!context) {
LOG("Failed to create context");
goto rollback_socket;
}
int events_fd = WindowGetEventsFd(context->window);
if (events_fd == -1) {
LOG("Failed to get events fd");
goto rollback_context;
}
int timer_fd = timerfd_create(CLOCK_MONOTONIC, 0);
if (timer_fd == -1) {
LOG("Failed to create timer (%s)", strerror(errno));
goto rollback_context;
}
static const unsigned ping_period_ns = 1000 * 1000 * 1000 / 3;
static const struct itimerspec spec = {
.it_interval.tv_nsec = ping_period_ns,
.it_value.tv_nsec = ping_period_ns,
};
if (timerfd_settime(timer_fd, 0, &spec, NULL)) {
LOG("Failed to arm timer (%s)", strerror(errno));
goto rollback_timer_fd;
}
if (signal(SIGINT, OnSignal) == SIG_ERR ||
signal(SIGTERM, OnSignal) == SIG_ERR) {
LOG("Failed to set signal handlers (%s)", strerror(errno));
goto rollback_timer_fd;
}
while (!g_signal) {
struct pollfd pfds[] = {
{.fd = sock, .events = POLLIN},
{.fd = events_fd, .events = POLLIN},
{.fd = timer_fd, .events = POLLIN},
};
switch (poll(pfds, LENGTH(pfds), -1)) {
case -1:
if (errno != EINTR) {
LOG("Failed to poll (%s)", strerror(errno));
goto rollback_timer_fd;
}
__attribute__((fallthrough));
case 0:
continue;
default:
break;
}
if (pfds[0].revents && !DemuxProtoStream(sock, context)) {
LOG("Failed to demux proto stream");
goto rollback_timer_fd;
}
if (pfds[1].revents && !WindowProcessEvents(context->window)) {
LOG("Failed to process window events");
goto rollback_timer_fd;
}
if (pfds[2].revents && !SendPingMessage(sock, timer_fd, context)) {
LOG("Failed to send ping message");
goto rollback_timer_fd;
}
}
rollback_timer_fd:
close(timer_fd);
rollback_context:
ContextDestroy(context);
rollback_socket:
close(sock);
bool result = g_signal == SIGINT || g_signal == SIGTERM;
return result ? EXIT_SUCCESS : EXIT_FAILURE;
}