/* * 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 "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; uint64_t timestamp; uint64_t ping_sum; uint64_t ping_count; uint64_t latency_sum; uint64_t 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), "Bitrate: %zu.000 Mbps", SIZE_MAX / 1000); *width = 4 + PuiStringWidth(str) + 4; *height = 4 + 12 * 3 + 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 bitrate_str[64]; uint64_t clock_delta = timestamp - context->timestamp; size_t bitrate = context->video_bitstream * 1000000 * 8 / clock_delta / 1024; snprintf(bitrate_str, sizeof(bitrate_str), "Bitrate: %zu.%03zu Mbps", bitrate / 1000, bitrate % 1000); 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 latency_str[64]; uint64_t latency = 0; if (context->latency_count) { // mburakov: Pessimistic calculations, these assume one fully missed vsync // for capture, one fully missed vsync for rendering, and 100Mbit network. latency = context->latency_sum / context->latency_count + ping + 16666 + 16666 + bitrate * 1000000 / 100000000 / context->latency_count; } snprintf(latency_str, sizeof(latency_str), "Latency: %zu.%03zu ms", latency / 1000, latency % 1000); size_t overlay_width = MAX(PuiStringWidth(bitrate_str), PuiStringWidth(ping_str)); overlay_width = MAX(overlay_width, PuiStringWidth(latency_str)) + 8; memset(buffer, 0, context->overlay_width * context->overlay_height * 4); for (size_t y = 0; y < context->overlay_height; y++) { for (size_t x = 0; x < overlay_width; x++) buffer[x + y * context->overlay_width] = 0x40000000; } PuiStringRender(bitrate_str, buffer + context->overlay_width * 4 + 4, context->overlay_width, 0xffffffff); PuiStringRender(ping_str, buffer + context->overlay_width * 16 + 4, context->overlay_width, 0xffffffff); PuiStringRender(latency_str, buffer + context->overlay_width * 28 + 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->latency_sum += proto->latency; context->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->timestamp = timestamp; context->ping_sum = 0; context->ping_count = 0; context->latency_sum = 0; context->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; } 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; }