From 6e554e50812879b72bb846bcacb99800f24211f9 Mon Sep 17 00:00:00 2001 From: Maxim Devaev Date: Fri, 17 Jul 2026 17:45:30 +0300 Subject: [PATCH] webcam: janus plugin --- janus/src/client.c | 124 +++++++++++----- janus/src/client.h | 42 +++--- janus/src/config.c | 14 +- janus/src/config.h | 2 + janus/src/h264.c | 269 +++++++++++++++++++++++++++++++++++ janus/src/h264.h | 28 ++++ janus/src/plugin.c | 288 ++++++++++++++++++++++++++++++++++--- janus/src/rtpc.c | 315 +++++++++++++++++++++++++++++++++++++++++ janus/src/rtpc.h | 82 +++++++++++ python/src/ustreamer.c | 11 +- src/libs/memsink.c | 176 ++++++++++++++++------- src/libs/memsink.h | 28 +++- src/libs/memsinksh.c | 15 +- src/libs/memsinksh.h | 12 +- 14 files changed, 1264 insertions(+), 142 deletions(-) create mode 100644 janus/src/h264.c create mode 100644 janus/src/h264.h create mode 100644 janus/src/rtpc.c create mode 100644 janus/src/rtpc.h diff --git a/janus/src/client.c b/janus/src/client.c index 0cd3036..8b0263f 100644 --- a/janus/src/client.c +++ b/janus/src/client.c @@ -38,11 +38,14 @@ #include "uslibs/array.h" #include "uslibs/list.h" #include "uslibs/ring.h" +#include "uslibs/frame.h" #include "au.h" #include "rtp.h" +static us_frame_s *_add_vplay_frame_to_ring(us_frame_s *frame, void *v_client); + static void *_video_thread(void *v_client); static void *_acap_thread(void *v_client); static void *_video_or_acap_thread(void *v_client, bool video); @@ -60,6 +63,7 @@ us_janus_client_s *us_janus_client_init(janus_callbacks *gw, janus_plugin_sessio atomic_init(&client->transmit, false); atomic_init(&client->transmit_acap, false); atomic_init(&client->transmit_aplay, false); + atomic_init(&client->transmit_vplay, false); atomic_init(&client->video_orient, 0); atomic_init(&client->stop, false); @@ -74,6 +78,8 @@ us_janus_client_s *us_janus_client_init(janus_callbacks *gw, janus_plugin_sessio US_RING_INIT_WITH_ITEMS(client->aplay_pcm_ring, 64, us_au_pcm_init); US_THREAD_CREATE(client->aplay_tid, _aplay_thread, client); + client->rtpc = us_rtpc_init(_add_vplay_frame_to_ring, client); + return client; } @@ -90,6 +96,9 @@ void us_janus_client_destroy(us_janus_client_s *client) { US_RING_DELETE_WITH_ITEMS(client->aplay_enc_ring, us_au_encoded_destroy); US_RING_DELETE_WITH_ITEMS(client->aplay_pcm_ring, us_au_pcm_destroy); + us_rtpc_destroy(client->rtpc); + // us_janus_client_stop_vplay(client); + free(client); } @@ -110,51 +119,100 @@ void us_janus_client_send(us_janus_client_s *client, const us_rtp_s *rtp) { } } +static us_frame_s *_add_vplay_frame_to_ring(us_frame_s *ready_frame, void *v_client) { + us_janus_client_s *const client = v_client; + + us_ring_s *const ring = client->vplay_enc_ring; + if (!ring) { + US_LOG_ERROR("Session %p has no vplay ring", client->session); + return ready_frame; + } + + const int ri = us_ring_producer_acquire(ring, 0); + if (ri < 0) { + US_LOG_ERROR("Session %p vplay ring is full", client->session); + return ready_frame; + } + + us_frame_s *const next_frame = ring->items[ri]; + ring->items[ri] = ready_frame; + us_ring_producer_release(ring, ri); + + return next_frame; +} + void us_janus_client_recv(us_janus_client_s *client, janus_plugin_rtp *packet) { - if ( - packet->video - || packet->length < sizeof(janus_rtp_header) - || !atomic_load(&client->transmit) - || !atomic_load(&client->transmit_aplay) - ) { + if (packet->length < sizeof(janus_rtp_header) || !atomic_load(&client->transmit)) { return; } + int payload_size = 0; + const u8 *const payload = (u8 *)janus_rtp_payload(packet->buffer, packet->length, &payload_size); + if (payload == NULL || payload_size < 1) { + return; + } const janus_rtp_header *const header = (janus_rtp_header*)packet->buffer; - if (header->type != US_RTP_OPUS_PAYLOAD) { - return; - } - const u16 seq = ntohs(header->seq_number); - if ( - seq >= client->aplay_seq_next // In order or missing - || (client->aplay_seq_next - seq) > 50 // In late sequence or sequence wrapped - ) { - client->aplay_seq_next = seq + 1; - int size = 0; - const char *const data = janus_rtp_payload(packet->buffer, packet->length, &size); - if (data == NULL || size <= 0) { - return; + if (packet->video) { + if (header->type == US_RTP_H264_PAYLOAD && atomic_load(&client->transmit_vplay)) { + const u32 rts = ntohl(header->timestamp); // RTS == RTP TimeStamp + bool retry; + do { + retry = false; + switch (us_rtpc_unwrap(client->rtpc, payload, payload_size, seq, rts)) { + case US_RUR_SUCCESS: + case US_RUR_BAD_PACKET: + case US_RUR_INVALID_SEQ: + case US_RUR_UNSUPPORTED_UNIT_TYPE: + case US_RUR_DEPACKETIZATION_SKIPPED: + break; + case US_RUR_DEPACKETIZATION_FAILED_RETRY: + retry = true; + // fall through + case US_RUR_DEPACKETIZATION_FAILED: + client->gw->send_pli(client->session); + break; + } + } while (retry); } - - us_ring_s *const ring = client->aplay_enc_ring; - const int ri = us_ring_producer_acquire(ring, 0); - if (ri < 0) { - // US_LOG_ERROR("Session %p aplay ring is full", client->session); - return; + } else { + if (header->type == US_RTP_OPUS_PAYLOAD && atomic_load(&client->transmit_aplay)) { + if (seq >= client->aplay_seq_next || (client->aplay_seq_next - seq) > 50) { + // ^^^ In order or missing |OR| ^^^ In late sequence or sequence wrapped + client->aplay_seq_next = seq + 1; + us_ring_s *const ring = client->aplay_enc_ring; + const int ri = us_ring_producer_acquire(ring, 0); + if (ri < 0) { + // US_LOG_ERROR("Session %p aplay ring is full", client->session); + return; + } + us_au_encoded_s *enc = ring->items[ri]; + if ((uz)payload_size < US_ARRAY_LEN(enc->data)) { + memcpy(enc->data, payload, payload_size); + enc->used = payload_size; + } else { + enc->used = 0; + } + us_ring_producer_release(ring, ri); + } } - us_au_encoded_s *enc = ring->items[ri]; - if ((uz)size < US_ARRAY_LEN(enc->data)) { - memcpy(enc->data, data, size); - enc->used = size; - } else { - enc->used = 0; - } - us_ring_producer_release(ring, ri); } } +void us_janus_client_start_vplay(us_janus_client_s *client, us_ring_s* vplay_enc_ring) { + if (atomic_exchange(&client->transmit_vplay, true)) { + return; + } + US_A(client->vplay_enc_ring == NULL); + client->vplay_enc_ring = vplay_enc_ring; +} + +/*void us_janus_client_stop_vplay(us_janus_client_s *client) { + atomic_store(&client->transmit_vplay, false); + client->vplay_enc_ring = NULL; +}*/ + static void *_video_thread(void *v_client) { US_THREAD_SETTLE("us_cx_vcap"); return _video_or_acap_thread(v_client, true); diff --git a/janus/src/client.h b/janus/src/client.h index 8e1f666..e535a00 100644 --- a/janus/src/client.h +++ b/janus/src/client.h @@ -32,32 +32,39 @@ #include "uslibs/ring.h" #include "rtp.h" +#include "rtpv.h" +#include "rtpc.h" typedef struct { janus_callbacks *gw; janus_plugin_session *session; - u32 video_ssrc; - u32 audio_ssrc; - atomic_bool transmit; - atomic_bool transmit_acap; - atomic_bool transmit_aplay; - atomic_uint video_orient; + u32 video_ssrc; + u32 audio_ssrc; - pthread_t video_tid; - pthread_t acap_tid; - pthread_t aplay_tid; - atomic_bool stop; + atomic_bool transmit; + atomic_bool transmit_acap; + atomic_bool transmit_vplay; + atomic_bool transmit_aplay; + atomic_uint video_orient; - us_ring_s *video_ring; - us_ring_s *acap_ring; + pthread_t video_tid; + pthread_t acap_tid; + pthread_t aplay_tid; + atomic_bool stop; - us_ring_s *aplay_enc_ring; - u16 aplay_seq_next; - us_ring_s *aplay_pcm_ring; + us_ring_s *video_ring; + us_ring_s *acap_ring; - US_LIST_DECLARE; + us_ring_s *aplay_enc_ring; + u16 aplay_seq_next; + us_ring_s *aplay_pcm_ring; + + us_rtpc_s *rtpc; + us_ring_s *vplay_enc_ring; + + US_LIST_DECLARE; } us_janus_client_s; @@ -66,3 +73,6 @@ void us_janus_client_destroy(us_janus_client_s *client); void us_janus_client_send(us_janus_client_s *client, const us_rtp_s *rtp); void us_janus_client_recv(us_janus_client_s *client, janus_plugin_rtp *packet); + +void us_janus_client_start_vplay(us_janus_client_s *client, us_ring_s* vplay_enc_ring); +// void us_janus_client_stop_vplay(us_janus_client_s *client); diff --git a/janus/src/config.c b/janus/src/config.c index c4bc7f7..85f3cc4 100644 --- a/janus/src/config.c +++ b/janus/src/config.c @@ -36,7 +36,7 @@ static char *_get_value(janus_config *jcfg, const char *section, const char *option); -static uint _get_uint(janus_config *jcfg, const char *section, const char *option, uint def); +static uint _get_uint(janus_config *jcfg, const char *section, const char *option, uint def, uint base); // static bool _get_bool(janus_config *jcfg, const char *section, const char *option, bool def); @@ -61,16 +61,20 @@ us_config_s *us_config_init(const char *config_dir_path) { US_LOG_ERROR("Missing config value: video.sink"); goto error; } + if ((config->acap_dev_name = _get_value(jcfg, "acap", "device")) != NULL) { - config->acap_hz = _get_uint(jcfg, "acap", "sampling_rate", 0); + config->acap_hz = _get_uint(jcfg, "acap", "sampling_rate", 0, 10); config->tc358743_dev_path = _get_value(jcfg, "acap", "tc358743"); if (config->acap_hz == 0 && !us_str_is_ok(config->tc358743_dev_path)) { US_LOG_ERROR("Either acap.sampling_rate or acap.tc358743 required"); goto error; } } + config->aplay_dev_name = _get_value(jcfg, "aplay", "device"); - // config->vplay_sink_name = _get_value(jcfg, "vplay", "sink"); + + config->vplay_sink_name = _get_value(jcfg, "vplay", "sink"); + config->vplay_sink_mode = _get_uint(jcfg, "vplay", "sink_mode", 0660, 8); goto ok; @@ -101,12 +105,12 @@ static char *_get_value(janus_config *jcfg, const char *section, const char *opt return us_strdup(option_obj->value); } -static uint _get_uint(janus_config *jcfg, const char *section, const char *option, uint def) { +static uint _get_uint(janus_config *jcfg, const char *section, const char *option, uint def, uint base) { char *const tmp = _get_value(jcfg, section, option); uint value = def; if (tmp != NULL) { errno = 0; - value = (uint)strtoul(tmp, NULL, 10); + value = (uint)strtoul(tmp, NULL, base); if (errno != 0) { value = def; } diff --git a/janus/src/config.h b/janus/src/config.h index d568fc1..59a116d 100644 --- a/janus/src/config.h +++ b/janus/src/config.h @@ -34,7 +34,9 @@ typedef struct { char *tc358743_dev_path; char *aplay_dev_name; + char *vplay_sink_name; + uint vplay_sink_mode; } us_config_s; diff --git a/janus/src/h264.c b/janus/src/h264.c new file mode 100644 index 0000000..6b1d350 --- /dev/null +++ b/janus/src/h264.c @@ -0,0 +1,269 @@ +/***************************************************************************** +# # +# uStreamer - Lightweight and fast MJPEG-HTTP streamer. # +# # +# This source file based on code of Janys-Gateway. # +# # +# Copyright (C) Lorenzo Miniero # +# Copyright (C) 2018-2024 Maxim Devaev # +# # +# This program 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. # +# # +# This program 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 this program. If not, see . # +# # +*****************************************************************************/ + + +#include "h264.h" + +#include "uslibs/types.h" +#include "uslibs/tools.h" + + +static INLINE u32 _eg_get_bit(const u8 *buf, u32 bit_offset) { + return ((*(buf + (bit_offset >> 3))) >> (7 - (bit_offset & 0x07))) & 0x01; +} + +static INLINE bool _is_valid_eg_offset(u32 bit_offset, uz size) { + return (bit_offset >> 3) + (bit_offset % 8 ? 1 : 0) < size; +} + +static bool _eg_skip_u1(const u8 *buf, u32 *bit_offset, uz size) { + (void)buf; + if (!_is_valid_eg_offset(*bit_offset, size)) { + return false; + } + ++(*bit_offset); + return true; +} + +static bool _eg_decode_u1(const u8 *buf, u32 *bit_offset, uz size, bool *out) { + if (!_is_valid_eg_offset(*bit_offset, size)) { + return false; + } + *out = (_eg_get_bit(buf, *bit_offset) != 0); + ++(*bit_offset); + return true; +} + +static bool _eg_skip_ev(const u8 *buf, u32 *bit_offset, uz size) { + u32 zeros = 0; + while (true) { + if (!_is_valid_eg_offset(*bit_offset, size)) { + return false; + } + if (_eg_get_bit(buf, (*bit_offset)++) != 0) { + break; + } + ++zeros; + } + for (; zeros; --zeros) { + if (!_is_valid_eg_offset((*bit_offset)++, size)) { + return false; + } + } + return true; +} + +static bool _eg_decode_uev(const u8 *buf, u32 *bit_offset, uint size, u32 *out) { + u32 zeros = 0; + while (true) { + if (!_is_valid_eg_offset(*bit_offset, size)) { + return false; + } + if (_eg_get_bit(buf, (*bit_offset)++) != 0) { + break; + } + ++zeros; + } + u32 res = (1 << zeros); + for (; zeros; --zeros) { + if (!_is_valid_eg_offset(*bit_offset, size)) { + return false; + } + res |= (_eg_get_bit(buf, *bit_offset) << (zeros - 1)); + ++(*bit_offset); + } + *out = res - 1; + return true; +} + +int us_h264_parse_sps_nalu(const u8 *buf, uz size, uint *width, uint *height) { + // NAL header + profile_idc + constraint_setX_flag, reserved_zero_2bits + level_idc + if (size < 4) { + return -1; + } + + // NAL header + ++buf; + --size; + + // profile_idc u(8) - Baseline profile + const u8 profile_idc = *buf; + if (profile_idc != 66) { + return -1; + } + ++buf; + --size; + + // constraint_set0_flag u(1) + // constraint_set1_flag u(1) + // constraint_set2_flag u(1) + // constraint_set3_flag u(1) + // constraint_set4_flag u(1) + // constraint_set5_flag u(1) + // reserved_zero_2bits u(2) + ++buf; + --size; + + // level_idc u(8) + ++buf; + --size; + + u32 bit_offset = 0; + + // seq_parameter_set_id ue(v) + if (!_eg_skip_ev(buf, &bit_offset, size)) { + return -1; + } + + // log2_max_frame_num_minus4 ue(v) + u32 log2_max_frame_num_minus4; + if (!_eg_decode_uev(buf, &bit_offset, size, &log2_max_frame_num_minus4)) { + return -1; + } + + // pic_order_cnt_type ue(v) + u32 pic_order_cnt_type; + if (!_eg_decode_uev(buf, &bit_offset, size, &pic_order_cnt_type)) { + return -1; + } + + if (pic_order_cnt_type == 0) { + // log2_max_pic_order_cnt_lsb_minus4 ue(v) + if (!_eg_skip_ev(buf, &bit_offset, size)) { + return -1; + } + } else if (pic_order_cnt_type == 1) { + // delta_pic_order_always_zero_flag u(1) + if (!_eg_skip_u1(buf, &bit_offset, size)) { + return -1; + } + + // offset_for_non_ref_pic se(v) + if (!_eg_skip_ev(buf, &bit_offset, size)) { + return -1; + } + + // offset_for_top_to_bottom_field se(v) + if (!_eg_skip_ev(buf, &bit_offset, size)) { + return -1; + } + + // num_ref_frames_in_pic_order_cnt_cycle ue(v) + u32 num_ref_frames_in_pic_order_cnt_cycle; + if (!_eg_decode_uev(buf, &bit_offset, size, &num_ref_frames_in_pic_order_cnt_cycle)) { + return -1; + } + + for (u32 i = 0; i < num_ref_frames_in_pic_order_cnt_cycle; ++i) { + // offset_for_ref_frame[i] se(v) + if (!_eg_skip_ev(buf, &bit_offset, size)) { + return -1; + } + } + } + + // max_num_ref_frames ue(v) + if (!_eg_skip_ev(buf, &bit_offset, size)) { + return -1; + } + + // gaps_in_frame_num_value_allowed_flag u(1) + if (!_eg_skip_u1(buf, &bit_offset, size)) { + return -1; + } + + // pic_width_in_mbs_minus1 ue(v) + u32 pic_width_in_mbs_minus1; + if (!_eg_decode_uev(buf, &bit_offset, size, &pic_width_in_mbs_minus1)) { + return -1; + } + + //pic_height_in_map_units_minus1 ue(v) + u32 pic_height_in_map_units_minus1; + if (!_eg_decode_uev(buf, &bit_offset, size, &pic_height_in_map_units_minus1)) { + return -1; + } + + // frame_mbs_only_flag u(1) + bool frame_mbs_only_flag; + if (!_eg_decode_u1(buf, &bit_offset, size, &frame_mbs_only_flag)) { + return -1; + } + + if (!frame_mbs_only_flag) { + // mb_adaptive_frame_field_flag u(1) + if (!_eg_skip_u1(buf, &bit_offset, size)) { + return -1; + } + } + + //direct_8x8_inference_flag u(1) + if (!_eg_skip_u1(buf, &bit_offset, size)) { + return -1; + } + + // frame_cropping_flag u(1) + bool frame_cropping_flag; + if (!_eg_decode_u1(buf, &bit_offset, size, &frame_cropping_flag)) { + return -1; + } + + u32 frame_crop_left_offset = 0; + u32 frame_crop_right_offset = 0; + u32 frame_crop_top_offset = 0; + u32 frame_crop_bottom_offset = 0; + if (frame_cropping_flag) { + // frame_crop_left_offset ue(v) + if (!_eg_decode_uev(buf, &bit_offset, size, &frame_crop_left_offset)) { + return -1; + } + + // frame_crop_right_offset ue(v) + if (!_eg_decode_uev(buf, &bit_offset, size, &frame_crop_right_offset)) { + return -1; + } + + // frame_crop_top_offset ue(v) + if (!_eg_decode_uev(buf, &bit_offset, size, &frame_crop_top_offset)) { + return -1; + } + + // frame_crop_bottom_offset ue(v) + if (!_eg_decode_uev(buf, &bit_offset, size, &frame_crop_bottom_offset)) { + return -1; + } + } + + *width = ( + (pic_width_in_mbs_minus1 + 1) * 16 + - frame_crop_left_offset * 2 + - frame_crop_right_offset * 2 + ); + *height = ( + (pic_height_in_map_units_minus1 + 1) * (2 - frame_mbs_only_flag) * 16 + - frame_crop_top_offset * 2 + - frame_crop_bottom_offset * 2 + ); + return 0; +} diff --git a/janus/src/h264.h b/janus/src/h264.h new file mode 100644 index 0000000..41e3008 --- /dev/null +++ b/janus/src/h264.h @@ -0,0 +1,28 @@ +/***************************************************************************** +# # +# uStreamer - Lightweight and fast MJPEG-HTTP streamer. # +# # +# Copyright (C) 2018-2024 Maxim Devaev # +# # +# This program 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. # +# # +# This program 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 this program. If not, see . # +# # +*****************************************************************************/ + + +#pragma once + +#include "uslibs/types.h" + + +int us_h264_parse_sps_nalu(const u8 *buf, uz size, uint *width, uint *height); diff --git a/janus/src/plugin.c b/janus/src/plugin.c index faff466..b7003a3 100644 --- a/janus/src/plugin.c +++ b/janus/src/plugin.c @@ -26,15 +26,13 @@ #include #include -#include -#include - #include #include #include #include #include #include +#include #include "uslibs/types.h" #include "uslibs/const.h" @@ -46,6 +44,7 @@ #include "uslibs/ring.h" #include "uslibs/memsink.h" #include "uslibs/chip.h" +#include "uslibs/memsink.h" #include "const.h" #include "client.h" @@ -58,7 +57,7 @@ #include "config.h" -static const char *const default_ice_url = "stun:stun.l.google.com:19302"; +static const char *const _g_default_ice_url = "stun:stun.l.google.com:19302"; static us_config_s *_g_config = NULL; static const useconds_t _g_watchers_polling = 100000; @@ -66,6 +65,7 @@ static const useconds_t _g_watchers_polling = 100000; static us_janus_client_s *_g_clients = NULL; static janus_callbacks *_g_gw = NULL; static us_ring_s *_g_video_ring = NULL; +static us_ring_s *_g_vplay_ring = NULL; static us_rtpv_s *_g_rtpv = NULL; static us_rtpa_s *_g_rtpa = NULL; @@ -73,12 +73,15 @@ static pthread_t _g_video_rtp_tid; static atomic_bool _g_video_rtp_tid_created = false; static pthread_t _g_video_sink_tid; static atomic_bool _g_video_sink_tid_created = false; +static pthread_t _g_vplay_tid; +static atomic_bool _g_vplay_tid_created = false; static pthread_t _g_acap_tid; static atomic_bool _g_acap_tid_created = false; static pthread_t _g_aplay_tid; static atomic_bool _g_aplay_tid_created = false; static pthread_mutex_t _g_video_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_mutex_t _g_vplay_lock = PTHREAD_MUTEX_INITIALIZER; static pthread_mutex_t _g_acap_lock = PTHREAD_MUTEX_INITIALIZER; static pthread_mutex_t _g_aplay_lock = PTHREAD_MUTEX_INITIALIZER; static atomic_bool _g_ready = false; @@ -88,18 +91,29 @@ static atomic_bool _g_has_listeners = false; static atomic_bool _g_has_speakers = false; static atomic_bool _g_key_required = false; +static struct { + bool active; + us_janus_client_s *client; + uint width; + uint height; + uint fps; +} _g_camera = {0}; // Access should be protected by _g_vplay_lock + #define _LOCK_VIDEO US_MUTEX_LOCK(_g_video_lock) #define _UNLOCK_VIDEO US_MUTEX_UNLOCK(_g_video_lock) +#define _LOCK_VPLAY US_MUTEX_LOCK(_g_vplay_lock) +#define _UNLOCK_VPLAY US_MUTEX_UNLOCK(_g_vplay_lock) + #define _LOCK_ACAP US_MUTEX_LOCK(_g_acap_lock) #define _UNLOCK_ACAP US_MUTEX_UNLOCK(_g_acap_lock) #define _LOCK_APLAY US_MUTEX_LOCK(_g_aplay_lock) #define _UNLOCK_APLAY US_MUTEX_UNLOCK(_g_aplay_lock) -#define _LOCK_ALL { _LOCK_VIDEO; _LOCK_ACAP; _LOCK_APLAY; } -#define _UNLOCK_ALL { _UNLOCK_APLAY; _UNLOCK_ACAP; _UNLOCK_VIDEO; } +#define _LOCK_ALL { _LOCK_VIDEO; _LOCK_VPLAY; _LOCK_ACAP; _LOCK_APLAY; } +#define _UNLOCK_ALL { _UNLOCK_APLAY; _UNLOCK_ACAP; _UNLOCK_VPLAY; _UNLOCK_VIDEO; } #define _READY atomic_load(&_g_ready) #define _STOP atomic_load(&_g_stop) @@ -372,6 +386,120 @@ static void *_aplay_thread(void *arg) { return NULL; } +static void _camera_set_active(uint width, uint height, uint fps); +static void _camera_set_inactive(); + +static void *_vplay_thread(void *arg) { + (void)arg; + US_THREAD_SETTLE("us_p_vplay"); + atomic_store(&_g_vplay_tid_created, true); + + us_frame_s *frame = us_frame_init(); + + while (!_STOP) { + us_memsink_s* sink = us_memsink_init_opened( + "H264-CAM", + _g_config->vplay_sink_name, + true, // Server + _g_config->vplay_sink_mode, + false, // Don't remote sink on destroy + 1, // Client TTL + 1); // Timeout + if (sink == NULL) { + goto close_memsink; + } + + const ldf memsink_check_interval = 1; + ldf last_memsink_check_time = 0; + + int once = 0; + while (!_STOP) { + if (!_g_camera.active) { + frame->used = 0; + } + + if (frame->used > 0 && (_g_camera.width != frame->width || _g_camera.height != frame->height)) { + US_ONCE({ US_LOG_INFO("Got WebRTC frame with wrong resolution"); }); + frame->used = 0; + } + + const ldf now_ts = us_get_now_monotonic(); + if (frame->used > 0 || last_memsink_check_time + memsink_check_interval < now_ts) { + last_memsink_check_time = now_ts; + + us_memsink_wants_s w_get = {0}; + switch (us_memsink_server_x_lock(sink, &w_get)) { + case US_MSS_SUCCESS: + if (w_get.format == V4L2_PIX_FMT_H264) { + if (w_get.width == _g_camera.width && w_get.height == _g_camera.height) { + if (us_memsink_server_x_is_consumed(sink)) { + if (!_g_camera.active) { + _camera_set_active(w_get.width, w_get.height, w_get.fps); + } else if (frame->used > 0) { + US_ONCE({ US_LOG_INFO("Streaming to the camera ..."); }); + us_memsink_server_x_put(sink, frame); + frame->used = 0; + } + } else { + if (!_g_camera.active) { + us_memsink_server_x_set_consumed(sink); + } + } + } else { + if (_g_camera.active) { + _camera_set_inactive(); + } + _camera_set_active(w_get.width, w_get.height, w_get.fps); + } + } else { + US_ONCE({ US_LOG_ERROR("Got invalid format from the camera: %u", w_get.format); }); + } + if (us_memsink_server_x_unlock(sink) < 0) { + goto close_memsink; + } + break; + + case US_MSS_NO_CLIENT: + if (_g_camera.active) { + _camera_set_inactive(); + } + break; + + case US_MSS_BUSY: // Busy by some client + break; + + case US_MSS_ERROR: + default: + goto close_memsink; + } + } + + if (frame->used == 0) { // Frame sent, get a new one from a client + const int in_ri = us_ring_consumer_acquire(_g_vplay_ring, 0.1); + if (in_ri >= 0) { + us_frame_s *tmp_frame = frame; + frame = _g_vplay_ring->items[in_ri]; + _g_vplay_ring->items[in_ri] = tmp_frame; + us_ring_consumer_release(_g_vplay_ring, in_ri); + } + } else { + if (_g_camera.active) { + US_ONCE({ US_LOG_INFO("No frames from WebRTC"); }); + } + usleep(1000); // Don't iterate too frequently + } + } + + close_memsink: + US_DELETE(sink, us_memsink_destroy); + US_LOG_INFO("Memsink closed"); + sleep(1); + } + + us_frame_destroy(frame); + return NULL; +} + static void _relay_rtp_clients(const us_rtp_s *rtp) { US_LIST_ITERATE(_g_clients, client, { us_janus_client_send(client, rtp); @@ -408,13 +536,21 @@ static int _plugin_init(janus_callbacks *gw, const char *config_dir_path) { US_RING_INIT_WITH_ITEMS(_g_video_ring, 64, us_frame_init); _g_rtpv = us_rtpv_init(_relay_rtp_clients); + + if (_g_config->vplay_sink_name != NULL) { + US_RING_INIT_WITH_ITEMS(_g_vplay_ring, 15, us_frame_init); + US_THREAD_CREATE(_g_vplay_tid, _vplay_thread, NULL); + } + if (_g_config->acap_dev_name != NULL) { _g_rtpa = us_rtpa_init(_relay_rtp_clients); US_THREAD_CREATE(_g_acap_tid, _acap_thread, NULL); } + if (_g_config->aplay_dev_name != NULL) { US_THREAD_CREATE(_g_aplay_tid, _aplay_thread, NULL); } + US_THREAD_CREATE(_g_video_rtp_tid, _video_rtp_thread, NULL); US_THREAD_CREATE(_g_video_sink_tid, _video_sink_thread, NULL); @@ -427,17 +563,22 @@ static void _plugin_destroy(void) { atomic_store(&_g_stop, true); # define JOIN(_tid) { if (atomic_load(&_tid##_created)) { US_THREAD_JOIN(_tid); } } + JOIN(_g_aplay_tid); + JOIN(_g_acap_tid); + JOIN(_g_vplay_tid); JOIN(_g_video_sink_tid); JOIN(_g_video_rtp_tid); - JOIN(_g_acap_tid); - JOIN(_g_aplay_tid); # undef JOIN US_LIST_ITERATE(_g_clients, client, { + if (_g_camera.client == client) { + _g_camera.client = NULL; + } US_LIST_REMOVE(_g_clients, client); us_janus_client_destroy(client); }); + US_RING_DELETE_WITH_ITEMS(_g_vplay_ring, us_frame_destroy); US_RING_DELETE_WITH_ITEMS(_g_video_ring, us_frame_destroy); US_DELETE(_g_rtpa, us_rtpa_destroy); @@ -447,6 +588,53 @@ static void _plugin_destroy(void) { US_LOGGING_DESTROY; } +static void _push_camera_event(us_janus_client_s *client, bool requested) { + json_t *const json_event_type = json_string(requested ? "requested" : "released"); + + json_t *const camera = json_object(); + json_object_set_new(camera, "action", json_event_type); + + json_t *const result = json_object(); + json_object_set_new(result, "ustreamer", json_string("event")); + json_object_set_new(result, "status", json_string("camera")); + json_object_set_new(result, "camera", camera); + + json_t *const event = json_object(); + json_object_set_new(event, "result", result); + + if (client) { + _g_gw->push_event(client->session, create(), NULL, event, NULL); + } else { + US_LIST_ITERATE(_g_clients, client, { + _g_gw->push_event(client->session, create(), NULL, event, NULL); + }); + } + + json_decref(event); +} + +static void _camera_set_active(uint width, uint height, uint fps) { + _LOCK_VPLAY; + US_A(!_g_camera.active); + _g_camera.active = true; + _g_camera.width = width; + _g_camera.height = height; + _g_camera.fps = fps; + _push_camera_event(_g_camera.client, true); + _UNLOCK_VPLAY; + US_LOG_INFO("Camera requested: %ux%u@%u", width, height, fps); +} + +static void _camera_set_inactive() { + _LOCK_VPLAY; + US_A(_g_camera.active); + _g_camera.active = false; + _g_camera.client = NULL; + _push_camera_event(NULL, false); + _UNLOCK_VPLAY; + US_LOG_INFO("Camera released"); +} + static void _plugin_create_session(janus_plugin_session *session, int *err) { _IF_DISABLED({ *err = -1; return; }); _LOCK_ALL; @@ -467,6 +655,9 @@ static void _plugin_destroy_session(janus_plugin_session* session, int *err) { US_LIST_ITERATE(_g_clients, client, { if (client->session == session) { US_LOG_INFO("Removing session %p ...", session); + if (_g_camera.client == client) { + _g_camera.client = NULL; // Is it required to notify other clients? + } US_LIST_REMOVE(_g_clients, client); us_janus_client_destroy(client); found = true; @@ -611,6 +802,8 @@ static struct janus_plugin_result *_plugin_handle_message( { _LOCK_ALL; + with_vplay = (with_vplay && _g_camera.active && !_g_camera.client); + bool has_listeners = false; bool has_speakers = false; US_LIST_ITERATE(_g_clients, client, { @@ -629,6 +822,10 @@ static struct janus_plugin_result *_plugin_handle_message( atomic_store(&client->transmit_acap, with_acap); atomic_store(&client->transmit_aplay, with_aplay); atomic_store(&client->video_orient, video_orient); + if (with_vplay) { + us_janus_client_start_vplay(client, _g_vplay_ring); + _g_camera.client = client; + } } has_listeners = (has_listeners || atomic_load(&client->transmit_acap)); has_speakers = (has_speakers || atomic_load(&client->transmit_aplay)); @@ -640,12 +837,31 @@ static struct janus_plugin_result *_plugin_handle_message( } else if (!strcmp(request_str, "features")) { const char *const ice_url = getenv("JANUS_USTREAMER_WEB_ICE_URL"); + + _LOCK_VPLAY + json_t *camera_req = NULL; + if (_g_camera.active) { + json_t *resolution = json_object(); + json_object_set_new(resolution, "width", json_integer(_g_camera.width)); + json_object_set_new(resolution, "height", json_integer(_g_camera.height)); + + camera_req = json_object(); + json_object_set_new(camera_req, "resolution", resolution); + json_object_set_new(camera_req, "fps", json_integer(_g_camera.fps)); + } + _UNLOCK_VPLAY + json_t *const features = json_pack( - "{s:b, s:b, s:{s:s?}}", + "{s:b, s:b, s:{s:b, s:o*}, s:{s:s?}}", "audio", us_au_probe(_g_config->acap_dev_name), "mic", us_au_probe(_g_config->aplay_dev_name), - "ice", "url", (ice_url != NULL ? ice_url : default_ice_url) + "camera", + "enabled", (_g_config->vplay_sink_name != NULL), + "request", camera_req, + "ice", + "url", (ice_url != NULL ? ice_url : _g_default_ice_url) ); + PUSH_STATUS("features", features, NULL); json_decref(features); @@ -672,30 +888,60 @@ done: static void _plugin_incoming_rtp(janus_plugin_session *session, janus_plugin_rtp *packet) { _IF_DISABLED({ return; }); - if (session == NULL || packet == NULL || packet->video) { - return; // Accept only valid audio + if (session == NULL || packet == NULL) { + return; // Accept only valid packets } - _LOCK_APLAY; - US_LIST_ITERATE(_g_clients, client, { - if (client->session == session) { - us_janus_client_recv(client, packet); - break; + if (packet->video) { + _LOCK_VPLAY; + if (_g_camera.client != NULL && _g_camera.client->session == session) { + us_janus_client_recv(_g_camera.client, packet); } - }); - _UNLOCK_APLAY; + _UNLOCK_VPLAY; + } else { + _LOCK_APLAY; + US_LIST_ITERATE(_g_clients, client, { + if (client->session == session) { + us_janus_client_recv(client, packet); + break; + } + }); + _UNLOCK_APLAY; + } } static void _plugin_incoming_rtcp(janus_plugin_session *session, janus_plugin_rtcp *packet) { _IF_DISABLED({ return; }); - if (session == NULL || packet == NULL || !packet->video) { - return; // Accept only valid video + if ( + session == NULL || packet == NULL || !packet->video + || packet->length < sizeof(janus_rtcp_header) + ) { + return; } + if ( janus_rtcp_has_pli(packet->buffer, packet->length) || janus_rtcp_has_fir(packet->buffer, packet->length) ) { atomic_store(&_g_key_required, true); } + + janus_rtcp_header *header = (janus_rtcp_header *)packet->buffer; + + if (header->type == RTCP_SR && janus_rtcp_check_sr(header, packet->length)) { + const janus_rtcp_sr *const sr = (janus_rtcp_sr *)header; + + if (sr->si.ntp_ts_msw && sr->si.ntp_ts_lsw) { + _LOCK_VPLAY; + US_LIST_ITERATE(_g_clients, client, { + if (client->session == session) { + const ldf ntp_ts = ntohl(sr->si.ntp_ts_msw) + ntohl(sr->si.ntp_ts_lsw) / 4294967296.L; + us_rtpc_sync_timestamp(client->rtpc, ntp_ts, ntohl(sr->si.rtp_ts)); + break; + } + }); + _UNLOCK_VPLAY; + } + } } diff --git a/janus/src/rtpc.c b/janus/src/rtpc.c new file mode 100644 index 0000000..f659069 --- /dev/null +++ b/janus/src/rtpc.c @@ -0,0 +1,315 @@ +/***************************************************************************** +# # +# uStreamer - Lightweight and fast MJPEG-HTTP streamer. # +# # +# Copyright (C) 2018-2024 Maxim Devaev # +# # +# This program 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. # +# # +# This program 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 this program. If not, see . # +# # +*****************************************************************************/ + + +#include "rtpc.h" + +#include + +#include "uslibs/logging.h" + +#include "h264.h" + + +// H.264 fragment types. Commented types are not used in WebRTC. +enum { + _FR_REGULAR_NAL_MIN = 1, + _FR_REGULAR_NAL_MAX = 23, + _FR_STAP_A = 24, // Single Time Aggregation Packet, Type A (STAP-A) + // _FR_STAP_B = 25, // Single Time Aggregation Packet, Type B (STAP-B) + // _FR_MTAP_16 = 26, // Multi-Time Aggregation Packet, Type A (MTAP-16) + // _FR_MTAP_24 = 27, // Multi-Time Aggregation Packet, Type B (MTAP-24) + _FR_FU_A = 28, // Fragmentation unit (FU-A) + // _FR_FU_B = 29, // Fragmentation unit (FU-B) +}; + +enum { + _NALU_SPS = 7 +}; + + +static int _update_sequence_for_new_packet(us_rtpc_s *rtpc, u16 seq); +static ldf _rtp_to_local_ts(us_rtpc_s *rtpc, u32 rts); +static void _update_timestamps_for_new_packet(us_rtpc_s *rtpc, u32 rts); + +static void _unwrapping_begin(us_rtpc_s *rtpc); +static void _unwrapping_end(us_rtpc_s *rtpc); +static void _unwrap_whole(us_rtpc_s *rtpc, const u8 *data, u16 size); + + +us_rtpc_s *us_rtpc_init(us_frame_callback_f callback, void *user_data) { + us_rtpc_s *rtpc; + US_CALLOC(rtpc, 1); + rtpc->user_data = user_data; + rtpc->frame = us_frame_init(); + rtpc->callback = callback; + return rtpc; +} + +void us_rtpc_destroy(us_rtpc_s *rtpc) { + us_frame_destroy(rtpc->frame); + free(rtpc); +} + +us_rtpc_unwrap_result_e us_rtpc_unwrap( + us_rtpc_s *rtpc, + const u8 *payload, + int payload_size, + u16 seq, + u32 rts +) { + US_A(payload != NULL); + US_A(payload_size >= 1); + + if (_update_sequence_for_new_packet(rtpc, seq) < 0) { + return US_RUR_INVALID_SEQ; + } + _update_timestamps_for_new_packet(rtpc, rts); + + rtpc->had_packets = true; + + const u8 fragment_type = payload[0] & 0x1F; + + if (rtpc->fu && (rtpc->fu_rts != rts || fragment_type != _FR_FU_A)) { + rtpc->fu = false; + if (!rtpc->fu_is_bad) { // Not reported yet + return US_RUR_DEPACKETIZATION_FAILED_RETRY; + } + } + + if (fragment_type >= _FR_REGULAR_NAL_MIN && fragment_type <= _FR_REGULAR_NAL_MAX) { + _unwrap_whole(rtpc, payload, payload_size); + + } else if (fragment_type == _FR_STAP_A) { // Single Time Aggregation Packet, Type A (STAP-A). + const u8 *aggregated_unit = payload + 1; // skip aggregation header (1 byte) + payload_size -= 1; + + while (payload_size > 0) { + if (payload_size < 2) { // Aggregated NAL unit size (2 bytes) is missing + US_LOG_ERROR("Aggregated NALU size is missing: seq=%u, rts=%u", seq, rts); + return US_RUR_BAD_PACKET; + } + + const u16 nalu_size = ntohs(*((const u16 *)aggregated_unit)); + aggregated_unit += 2; + payload_size -= 2; + + if (nalu_size == 0) { + US_LOG_ERROR("Aggregated NALU size is zero: seq=%u, rts=%u", seq, rts); + return US_RUR_BAD_PACKET; + } + + if (payload_size < nalu_size) { // aggregated unit size declared as bigger than available data size + US_LOG_ERROR("Aggregated NALU size is too big: seq=%u, rts=%u", seq, rts); + return US_RUR_BAD_PACKET; + } + + _unwrap_whole(rtpc, aggregated_unit, nalu_size); + + aggregated_unit += nalu_size; + payload_size -= nalu_size; + } + + } else if (fragment_type == _FR_FU_A) { // Fragmentation unit (FU-A). + if (payload_size < 2) { // FU header is missing (2 bytes) + US_LOG_ERROR("FU header is missing: seq=%u, rts=%u", seq, rts); + return US_RUR_BAD_PACKET; + } + + const u8 fu_indicator = payload[0]; + const u8 fu_header = payload[1]; + const bool first_fragment = !!(fu_header & 0x80); + const bool last_fragment = !!(fu_header & 0x40); + + if (first_fragment && last_fragment) { // should never happen according to RFC + return US_RUR_BAD_PACKET; + } + + if (rtpc->fu && rtpc->fu_is_bad) { + return US_RUR_DEPACKETIZATION_SKIPPED; + } + + if (!rtpc->fu) { + rtpc->fu = true; + rtpc->fu_seq = seq; + rtpc->fu_rts = rts; + + if (!first_fragment) { + // US_LOG_INFO("Ignoring FU without a first fragment: seq=%u, rts=%u", seq, rts); + rtpc->fu_is_bad = true; + return US_RUR_DEPACKETIZATION_FAILED; + } + rtpc->fu_is_bad = false; + + const u8 header = (fu_indicator & 0xE0) | (fu_header & 0x1F); + _unwrapping_begin(rtpc); + us_frame_append_data(rtpc->frame, &header, sizeof(header)); + // - FU indicator - FU header + us_frame_append_data(rtpc->frame, (const u8 *)payload + 2, payload_size - 2); + + } else if (!first_fragment && (seq == rtpc->fu_seq + 1) && (rts == rtpc->fu_rts)) { + // ^^^ Already had first fragment + rtpc->fu_seq = seq; + + // - FU indicator - FU header + us_frame_append_data(rtpc->frame, (const u8 *)payload + 2, payload_size - 2); + + if (last_fragment) { + _unwrapping_end(rtpc); + rtpc->fu = false; + } + + } else { + // US_LOG_ERROR("Something went wrong with FU processing: seq=%u, rts=%u", seq, rts); + rtpc->fu_is_bad = true; + return US_RUR_DEPACKETIZATION_FAILED; + } + + } else { + US_LOG_ERROR("Unsupported video unit type=%x: seq=%u, rts=%u", fragment_type, seq, rts); + return US_RUR_UNSUPPORTED_UNIT_TYPE; + } + return US_RUR_SUCCESS; +} + +void us_rtpc_sync_timestamp(us_rtpc_s *rtpc, ldf ntp_ts, u32 packet_rts) { + if (!rtpc->had_packets) { + return; + } + if (rtpc->reference_ntp_ts > ntp_ts) { + return; + } + + if (rtpc->reference_ntp_ts == 0.L) { + if (rtpc->reference_packet_rts < packet_rts) { + const ldf diff = (packet_rts - rtpc->reference_packet_rts) / 90000.L; + rtpc->reference_local_ts += diff; + rtpc->reference_packet_rts = packet_rts; + rtpc->reference_ntp_ts = ntp_ts; + } + } else { + const ldf diff = ntp_ts - rtpc->reference_ntp_ts; + rtpc->reference_local_ts += diff; + rtpc->reference_packet_rts = packet_rts; + rtpc->reference_ntp_ts = ntp_ts; + } +} + +static int _update_sequence_for_new_packet(us_rtpc_s *rtpc, u16 seq) { + if ( + rtpc->had_packets + && ( + rtpc->last_packet_seq == seq + || (rtpc->last_packet_seq < seq && seq - rtpc->last_packet_seq > (UINT16_MAX >> 1)) + || (seq < rtpc->last_packet_seq && rtpc->last_packet_seq - seq < (UINT16_MAX >> 1)) + ) + ) { + return -1; + } + rtpc->last_packet_seq = seq; + return 0; +} + +static ldf _rtp_to_local_ts(us_rtpc_s *rtpc, u32 rts) { + if (rtpc->reference_packet_rts <= rts) { + return rtpc->reference_local_ts + (rts - rtpc->reference_packet_rts) / 90000.L; + } else { + return rtpc->reference_local_ts - (rtpc->reference_packet_rts - rts) / 90000.L; + } +} + +static void _update_timestamps_for_new_packet(us_rtpc_s *rtpc, u32 rts) { + const ldf now_ts = us_get_now_monotonic(); + ldf ts; // Fill for any goto + + if (!rtpc->had_packets) { + ts = _rtp_to_local_ts(rtpc, rts); // Used in packet_update + goto full_update; + } + if (rtpc->last_packet_rts != rts) { // If this is a new frame + ts = _rtp_to_local_ts(rtpc, rts); + if (rtpc->last_packet_rts > rts || fabsl(now_ts - ts) > .1L) { + // ^ RTP timestamp wrapped |OR| ^ Too significant mistake + goto full_update; + } + goto packet_update; + } + + return; + +full_update: + rtpc->reference_local_ts = now_ts; + rtpc->reference_packet_rts = rts; + rtpc->reference_ntp_ts = 0.L; // Force resync + // US_LOG_INFO("Reference timestamps reseted"); +packet_update: + rtpc->last_packet_rts = rts; + rtpc->frame_ts = ts; +} + +static void _unwrapping_begin(us_rtpc_s *rtpc) { + us_frame_s *const frame = rtpc->frame; + u8 *const data = frame->data; + const uz allocated = frame->allocated; + + // Reset the content, start from scratch + memset(frame, 0, sizeof(us_frame_s)); + frame->data = data; + frame->allocated = allocated; + + static const u8 nalu_start_code[] = {0, 0, 0, 1}; + us_frame_set_data(rtpc->frame, nalu_start_code, sizeof(nalu_start_code)); +} + +static void _unwrapping_end(us_rtpc_s *rtpc) { + us_frame_s *const frame = rtpc->frame; + frame->format = V4L2_PIX_FMT_H264; + frame->width = rtpc->frame_width; + frame->height = rtpc->frame_height; + frame->grab_begin_ts = rtpc->frame_ts; + frame->grab_end_ts = rtpc->frame_ts; + + // Exchange frames + rtpc->frame = rtpc->callback(frame, rtpc->user_data); + US_A(rtpc->frame != NULL); +} + +static void _unwrap_whole(us_rtpc_s *rtpc, const u8 *data, u16 size) { + US_A(size >= 1); + + const u8 nalu_type = (data[0] & 0x1F); + if (nalu_type == _NALU_SPS) { + uint width = 0; + uint height = 0; + if ( + !us_h264_parse_sps_nalu(data, size, &width, &height) + && (rtpc->frame_width != width || rtpc->frame_height != height) + ) { + US_LOG_INFO("Frame size updated: width=%u, height=%u", width, height); + rtpc->frame_width = width; + rtpc->frame_height = height; + } + } + + _unwrapping_begin(rtpc); + us_frame_append_data(rtpc->frame, data, size); + _unwrapping_end(rtpc); +} diff --git a/janus/src/rtpc.h b/janus/src/rtpc.h new file mode 100644 index 0000000..9d12e07 --- /dev/null +++ b/janus/src/rtpc.h @@ -0,0 +1,82 @@ +/***************************************************************************** +# # +# uStreamer - Lightweight and fast MJPEG-HTTP streamer. # +# # +# Copyright (C) 2018-2024 Maxim Devaev # +# # +# This program 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. # +# # +# This program 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 this program. If not, see . # +# # +*****************************************************************************/ + + +#pragma once + +#include + +#include "uslibs/types.h" +#include "uslibs/frame.h" + + +// Should return: +// * input frame +// * replacement frame - to use it in following parsing +typedef us_frame_s *(*us_frame_callback_f)(us_frame_s *frame, void *user_data); + +typedef struct { // *_rts == [R]TP [T]ime[S]tamp + us_frame_callback_f callback; + void *user_data; + + bool had_packets; // Были ли вообще какие-то пакеты когда-то, взводится один раз + + ldf reference_local_ts; // Локальное время + u32 reference_packet_rts; + ldf reference_ntp_ts; + + u16 last_packet_seq; + u32 last_packet_rts; // Время из последнего полученного пакета + + bool fu; + u16 fu_seq; + u32 fu_rts; + bool fu_is_bad; + + us_frame_s *frame; + uint frame_width; + uint frame_height; + ldf frame_ts; // in case of fu, time of very first packet +} us_rtpc_s; + +typedef enum { + US_RUR_SUCCESS = 0, + US_RUR_BAD_PACKET = -1, + US_RUR_INVALID_SEQ = -2, + US_RUR_DEPACKETIZATION_FAILED = -3, + // Incomplete FU found. Current packet not handled. + // It's required to call us_rtpc_unwrap one else time with the same packet. + US_RUR_DEPACKETIZATION_FAILED_RETRY = -4, + US_RUR_DEPACKETIZATION_SKIPPED = -5, + US_RUR_UNSUPPORTED_UNIT_TYPE = -6, +} us_rtpc_unwrap_result_e; + +us_rtpc_s *us_rtpc_init(us_frame_callback_f callback, void *user_data); +void us_rtpc_destroy(us_rtpc_s *rtpc); + +us_rtpc_unwrap_result_e us_rtpc_unwrap( + us_rtpc_s *rtpc, + const u8 *payload, + int payload_size, + u16 seq, + u32 rts); + +void us_rtpc_sync_timestamp(us_rtpc_s *rtpc, ldf ntp_ts, u32 packet_rts); diff --git a/python/src/ustreamer.c b/python/src/ustreamer.c index e5fd8c4..0850a8e 100644 --- a/python/src/ustreamer.c +++ b/python/src/ustreamer.c @@ -63,7 +63,7 @@ typedef struct { static void _MemsinkObject_destroy_internals(_MemsinkObject *self) { if (self->mem != NULL) { - us_memsink_shared_unmap(self->mem, self->data_size); + us_memsinksh_unmap(self->mem, self->data_size); self->mem = NULL; } US_CLOSE_FD(self->fd); @@ -94,7 +94,7 @@ static int _MemsinkObject_init(_MemsinkObject *self, PyObject *args, PyObject *k SET_DOUBLE(drop_same_frames, >= 0); # undef SET_DOUBLE - if ((self->data_size = us_memsink_calculate_size(self->obj)) == 0) { + if ((self->data_size = us_memsinksh_calculate_size(self->obj)) == 0) { PyErr_SetString(PyExc_ValueError, "Invalid memsink object suffix"); return -1; } @@ -105,7 +105,7 @@ static int _MemsinkObject_init(_MemsinkObject *self, PyObject *args, PyObject *k PyErr_SetFromErrno(PyExc_OSError); goto error; } - if ((self->mem = us_memsink_shared_map(self->fd, self->data_size)) == NULL) { + if ((self->mem = us_memsinksh_map(self->fd, self->data_size)) == NULL) { PyErr_SetFromErrno(PyExc_OSError); goto error; } @@ -165,6 +165,7 @@ static int _wait_frame(_MemsinkObject *self) { // Let the sink know that the client is alive mem->last_client_ts = now_ts; + mem->client_magic = US_MEMSINK_MAGIC; if (mem->id == self->frame_id) { goto retry; @@ -174,7 +175,7 @@ static int _wait_frame(_MemsinkObject *self) { if ( US_FRAME_COMPARE_GEOMETRY(self->mem, self->frame) && (self->frame_ts + self->drop_same_frames > now_ts) - && !memcmp(self->frame->data, us_memsink_get_data(mem), mem->used) + && !memcmp(self->frame->data, us_memsinksh_get_data(mem), mem->used) ) { self->frame_id = mem->id; goto retry; @@ -225,7 +226,7 @@ static PyObject *_MemsinkObject_wait_frame(_MemsinkObject *self, PyObject *args, } us_memsink_shared_s *mem = self->mem; - us_frame_set_data(self->frame, us_memsink_get_data(mem), mem->used); + us_frame_set_data(self->frame, us_memsinksh_get_data(mem), mem->used); US_FRAME_COPY_META(self->mem, self->frame); self->frame_id = mem->id; self->frame_ts = us_get_now_monotonic(); diff --git a/src/libs/memsink.c b/src/libs/memsink.c index 1150a1d..15d83bf 100644 --- a/src/libs/memsink.c +++ b/src/libs/memsink.c @@ -56,7 +56,7 @@ us_memsink_s *us_memsink_init_opened( US_LOG_INFO("Using %s-sink: %s", name, obj); - if ((sink->data_size = us_memsink_calculate_size(obj)) == 0) { + if ((sink->data_size = us_memsinksh_calculate_size(obj)) == 0) { US_LOG_ERROR("%s-sink: Invalid object suffix", name); goto error; } @@ -75,7 +75,7 @@ us_memsink_s *us_memsink_init_opened( goto error; } - if ((sink->mem = us_memsink_shared_map(sink->fd, sink->data_size)) == NULL) { + if ((sink->mem = us_memsinksh_map(sink->fd, sink->data_size)) == NULL) { US_LOG_PERROR("%s-sink: Can't mmap shared memory", name); goto error; } @@ -88,7 +88,7 @@ error: void us_memsink_destroy(us_memsink_s *sink) { if (sink->mem != NULL) { - if (us_memsink_shared_unmap(sink->mem, sink->data_size) < 0) { + if (us_memsinksh_unmap(sink->mem, sink->data_size) < 0) { US_LOG_PERROR("%s-sink: Can't unmap shared memory", sink->name); } } @@ -143,11 +143,10 @@ bool us_memsink_server_check(us_memsink_s *sink, const us_frame_s *frame) { } // Проверяем, есть ли у нас живой клиент по таймауту - const bool has_clients = (sink->mem->last_client_ts + sink->client_ttl > us_get_now_monotonic()); + const bool has_clients = us_memsinksh_has_clients(sink->mem, sink->client_ttl); atomic_store(&sink->has_clients, has_clients); - if (flock(sink->fd, LOCK_UN) < 0) { - US_LOG_PERROR("%s-sink: Can't unlock memory", sink->name); + if (us_memsink_server_x_unlock(sink) < 0) { return false; } if (has_clients) { @@ -163,45 +162,34 @@ bool us_memsink_server_check(us_memsink_s *sink, const us_frame_s *frame) { int us_memsink_server_put( us_memsink_s *sink, const us_frame_s *frame, - us_memsink_wants_s *wants + us_memsink_wants_s *w_get ) { US_A(sink->server); const ldf now = us_get_now_monotonic(); - if (frame->used > sink->data_size) { - US_LOG_ERROR("%s-sink: Can't put frame: is too big (%zu > %zu)", - sink->name, frame->used, sink->data_size); - return 0; - } - if (us_flock_timedwait_monotonic(sink->fd, 1) == 0) { - US_LOG_VERBOSE("%s-sink: >>>>> Exposing new frame ...", sink->name); - - sink->mem->id = us_get_now_id(); - if (sink->mem->wants.key && frame->key) { - sink->mem->wants.key = false; - } - if (wants != NULL) { - memcpy(wants, &sink->mem->wants, sizeof(us_memsink_wants_s)); - } - - memcpy(us_memsink_get_data(sink->mem), frame->data, frame->used); - sink->mem->used = frame->used; - US_FRAME_COPY_META(frame, sink->mem); - - sink->mem->magic = US_MEMSINK_MAGIC; - sink->mem->version = US_MEMSINK_VERSION; - - const bool has_clients = (sink->mem->last_client_ts + sink->client_ttl > us_get_now_monotonic()); + const bool has_clients = us_memsinksh_has_clients(sink->mem, sink->client_ttl); atomic_store(&sink->has_clients, has_clients); - if (flock(sink->fd, LOCK_UN) < 0) { - US_LOG_PERROR("%s-sink: Can't unlock memory", sink->name); + US_LOG_VERBOSE("%s-sink: >>>>> Exposing new frame ...", sink->name); + const bool exposed = !us_memsink_server_x_put(sink, frame); + + if (w_get != NULL) { + if (has_clients) { + memcpy(w_get, &sink->mem->wants, sizeof(us_memsink_wants_s)); + } else { + US_MEMSET_ZERO(*w_get); + } + } + + if (us_memsink_server_x_unlock(sink) < 0) { return -1; } - US_LOG_VERBOSE("%s-sink: Exposed new frame; full exposition time = %.3Lf", - sink->name, us_get_now_monotonic() - now); + if (exposed) { + US_LOG_VERBOSE("%s-sink: Exposed new frame; full exposition time = %.3Lf", + sink->name, us_get_now_monotonic() - now); + } } else if (errno == EWOULDBLOCK) { US_LOG_VERBOSE("%s-sink: ===== Shared memory is busy now; frame skipped", sink->name); @@ -216,8 +204,8 @@ int us_memsink_server_put( int us_memsink_client_get( us_memsink_s *sink, us_frame_s *frame, - us_memsink_wants_s *get, - const us_memsink_wants_s *put + us_memsink_wants_s *w_get, + const us_memsink_wants_s *w_put ) { US_A(!sink->server); // Client only @@ -242,33 +230,121 @@ int us_memsink_client_get( goto done; } + const bool had_client_magic = (sink->mem->client_magic == US_MEMSINK_MAGIC); // Let the sink know that the client is alive sink->mem->last_client_ts = us_get_now_monotonic(); + sink->mem->client_magic = US_MEMSINK_MAGIC; - if (sink->mem->id == sink->last_readed_id) { - retval = US_ERROR_NO_DATA; // Not updated + if (sink->mem->used == 0 || sink->mem->id == sink->last_readed_id) { + // ^ Zero frame |OR| ^ Not updated + retval = US_ERROR_NO_DATA; goto done; } sink->last_readed_id = sink->mem->id; - us_frame_set_data(frame, us_memsink_get_data(sink->mem), sink->mem->used); + us_frame_set_data(frame, us_memsinksh_get_data(sink->mem), sink->mem->used); US_FRAME_COPY_META(sink->mem, frame); - if (get != NULL) { - memcpy(get, &sink->mem->wants, sizeof(us_memsink_wants_s)); - } - if (put != NULL) { - const bool key = sink->mem->wants.key; - memcpy(&sink->mem->wants, put, sizeof(us_memsink_wants_s)); - if (key) { - sink->mem->wants.key = key; + if (w_get != NULL) { + if (had_client_magic) { // Чтобы не прочитать мусор, если не было клиентов + memcpy(w_get, &sink->mem->wants, sizeof(us_memsink_wants_s)); + } else { + US_MEMSET_ZERO(*w_get); } } + if (w_put != NULL) { + const bool key = sink->mem->wants.key; + memcpy(&sink->mem->wants, w_put, sizeof(us_memsink_wants_s)); + if (key) { + sink->mem->wants.key = key; + } + } else { + US_MEMSET_ZERO(sink->mem->wants); + } + done: - if (flock(sink->fd, LOCK_UN) < 0) { - US_LOG_PERROR("%s-sink: Can't unlock memory", sink->name); + if (us_memsink_server_x_unlock(sink) < 0) { retval = -1; } return retval; } + +us_mss_lock_result_e us_memsink_server_x_lock( + us_memsink_s *sink, + us_memsink_wants_s *w_get +) { + US_A(sink->server); + + if (us_flock_timedwait_monotonic(sink->fd, 1) < 0) { + if (errno == EWOULDBLOCK) { + return US_MSS_BUSY; + } + US_LOG_PERROR("%s-sink: Can't lock memory", sink->name); + return US_MSS_ERROR; + } + + if (sink->mem->magic == US_MEMSINK_MAGIC && sink->mem->version == US_MEMSINK_VERSION) { + // Если инициализировано - значит клиенту что-то уже могли написать + const bool has_clients = us_memsinksh_has_clients(sink->mem, sink->client_ttl); + atomic_store(&sink->has_clients, has_clients); + if (has_clients) { + if (w_get != NULL) { + memcpy(w_get, &sink->mem->wants, sizeof(us_memsink_wants_s)); + } + // sink->unsafe_last_client_ts = sink->mem->last_client_ts; // _x_ functions don't use it + return US_MSS_SUCCESS; + } + } else { + // А иначе скормим клиентам инициализацию с пустым фреймом, + // в следующий раз прочтем от них что-то путное. + us_memsink_server_x_put(sink, NULL); + } + + if (us_memsink_server_x_unlock(sink) < 0) { + return US_MSS_ERROR; + } + return US_MSS_NO_CLIENT; +} + +bool us_memsink_server_x_is_consumed(us_memsink_s *sink) { + return (sink->mem->used == 0); +} + +void us_memsink_server_x_set_consumed(us_memsink_s *sink) { + sink->mem->used = 0; +} + +int us_memsink_server_x_put(us_memsink_s *sink, const us_frame_s *frame) { + if (frame != NULL) { + if (frame->used > sink->data_size) { + US_LOG_ERROR("%s-sink: Can't put frame: it's too big (%zu > %zu)", + sink->name, frame->used, sink->data_size); + return -1; + } + + memcpy(us_memsinksh_get_data(sink->mem), frame->data, frame->used); + sink->mem->used = frame->used; + US_FRAME_COPY_META(frame, sink->mem); + + if (sink->mem->wants.key && frame->key) { + // Можем писать в мусор, но пофигу + sink->mem->wants.key = false; + } + } else { + sink->mem->used = 0; + } + + sink->mem->id = us_get_now_id(); + sink->mem->version = US_MEMSINK_VERSION; + sink->mem->magic = US_MEMSINK_MAGIC; + return 0; +} + +int us_memsink_server_x_unlock(us_memsink_s *sink) { + if (flock(sink->fd, LOCK_UN) < 0) { + US_LOG_PERROR("%s-sink: Can't unlock memory", sink->name); + return -1; + } + return 0; +} diff --git a/src/libs/memsink.h b/src/libs/memsink.h index 224bfe6..62652a8 100644 --- a/src/libs/memsink.h +++ b/src/libs/memsink.h @@ -61,10 +61,32 @@ bool us_memsink_server_check(us_memsink_s *sink, const us_frame_s *frame); int us_memsink_server_put( us_memsink_s *sink, const us_frame_s *frame, - us_memsink_wants_s *wants); + us_memsink_wants_s *w_get); int us_memsink_client_get( us_memsink_s *sink, us_frame_s *frame, - us_memsink_wants_s *get, - const us_memsink_wants_s *put); + us_memsink_wants_s *w_get, + const us_memsink_wants_s *w_put); + + +// Low-level + +typedef enum { + US_MSS_SUCCESS = 0, + US_MSS_ERROR = -1, + US_MSS_BUSY = -2, + US_MSS_NO_CLIENT = -3, +} us_mss_lock_result_e; + + +us_mss_lock_result_e us_memsink_server_x_lock( + us_memsink_s *sink, + us_memsink_wants_s *w_get); + +bool us_memsink_server_x_is_consumed(us_memsink_s *sink); +void us_memsink_server_x_set_consumed(us_memsink_s *sink); + +int us_memsink_server_x_put(us_memsink_s *sink, const us_frame_s *frame); + +int us_memsink_server_x_unlock(us_memsink_s *sink); diff --git a/src/libs/memsinksh.c b/src/libs/memsinksh.c index d0314a9..47725fd 100644 --- a/src/libs/memsinksh.c +++ b/src/libs/memsinksh.c @@ -31,7 +31,7 @@ #include "tools.h" -us_memsink_shared_s *us_memsink_shared_map(int fd, uz data_size) { +us_memsink_shared_s *us_memsinksh_map(int fd, uz data_size) { us_memsink_shared_s *mem = mmap( NULL, sizeof(us_memsink_shared_s) + data_size, @@ -44,12 +44,12 @@ us_memsink_shared_s *us_memsink_shared_map(int fd, uz data_size) { return mem; } -int us_memsink_shared_unmap(us_memsink_shared_s *mem, uz data_size) { +int us_memsinksh_unmap(us_memsink_shared_s *mem, uz data_size) { US_A(mem != NULL); return munmap(mem, sizeof(us_memsink_shared_s) + data_size); } -uz us_memsink_calculate_size(const char *obj) { +uz us_memsinksh_calculate_size(const char *obj) { const char *ptr = strrchr(obj, ':'); if (ptr == NULL) { ptr = strrchr(obj, '.'); @@ -67,6 +67,13 @@ uz us_memsink_calculate_size(const char *obj) { return 0; } -u8 *us_memsink_get_data(us_memsink_shared_s *mem) { +u8 *us_memsinksh_get_data(us_memsink_shared_s *mem) { return (u8*)(mem) + sizeof(us_memsink_shared_s); } + +bool us_memsinksh_has_clients(us_memsink_shared_s *mem, u32 ttl) { + return ( + mem->client_magic == US_MEMSINK_MAGIC + && mem->last_client_ts + ttl > us_get_now_monotonic() + ); +} diff --git a/src/libs/memsinksh.h b/src/libs/memsinksh.h index 80ac6ba..9906c46 100644 --- a/src/libs/memsinksh.h +++ b/src/libs/memsinksh.h @@ -27,7 +27,7 @@ #define US_MEMSINK_MAGIC ((u64)0xCAFEBABECAFEBABE) -#define US_MEMSINK_VERSION ((u32)10) +#define US_MEMSINK_VERSION ((u32)11) typedef struct { @@ -44,6 +44,7 @@ typedef struct { u64 id; uz used; + u64 client_magic; ldf last_client_ts; us_memsink_wants_s wants; @@ -51,8 +52,9 @@ typedef struct { } us_memsink_shared_s; -us_memsink_shared_s *us_memsink_shared_map(int fd, uz data_size); -int us_memsink_shared_unmap(us_memsink_shared_s *mem, uz data_size); +us_memsink_shared_s *us_memsinksh_map(int fd, uz data_size); +int us_memsinksh_unmap(us_memsink_shared_s *mem, uz data_size); -uz us_memsink_calculate_size(const char *obj); -u8 *us_memsink_get_data(us_memsink_shared_s *mem); +uz us_memsinksh_calculate_size(const char *obj); +u8 *us_memsinksh_get_data(us_memsink_shared_s *mem); +bool us_memsinksh_has_clients(us_memsink_shared_s *mem, u32 ttl);