forked/ustreamer
1
0
Fork 0
mirror of https://github.com/pikvm/ustreamer.git synced 2026-08-31 23:31:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

webcam: janus plugin

Browse Source
This commit is contained in:
Maxim Devaev
2026-07-17 17:45:30 +03:00
parent b5e12a8412
commit 6e554e5081
14 changed files with 1264 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
janus/src/client.c
View File

@@ -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);

42
janus/src/client.h
View File

@@ -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);

14
janus/src/config.c
View File

@@ -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;
}

2
janus/src/config.h
View File

@@ -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;

269
janus/src/h264.c Normal file
View File

@@ -0,0 +1,269 @@
/*****************************************************************************
# #
# uStreamer - Lightweight and fast MJPEG-HTTP streamer. #
# #
# This source file based on code of Janys-Gateway. #
# #
# Copyright (C) Lorenzo Miniero <lorenzo@meetecho.com> #
# Copyright (C) 2018-2024 Maxim Devaev <mdevaev@gmail.com> #
# #
# 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 <https://www.gnu.org/licenses/>. #
# #
*****************************************************************************/
#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;
}

28
janus/src/h264.h Normal file
View File

@@ -0,0 +1,28 @@
/*****************************************************************************
# #
# uStreamer - Lightweight and fast MJPEG-HTTP streamer. #
# #
# Copyright (C) 2018-2024 Maxim Devaev <mdevaev@gmail.com> #
# #
# 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 <https://www.gnu.org/licenses/>. #
# #
*****************************************************************************/
#pragma once
#include "uslibs/types.h"
int us_h264_parse_sps_nalu(const u8 *buf, uz size, uint *width, uint *height);

288
janus/src/plugin.c
View File

@@ -26,15 +26,13 @@
#include <fcntl.h>
#include <errno.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <pthread.h>
#include <jansson.h>
#include <janus/plugins/plugin.h>
#include <janus/rtp.h>
#include <janus/rtcp.h>
#include <alsa/asoundlib.h>
#include <linux/videodev2.h>
#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;
}
}
}

315
janus/src/rtpc.c Normal file
View File

@@ -0,0 +1,315 @@
/*****************************************************************************
# #
# uStreamer - Lightweight and fast MJPEG-HTTP streamer. #
# #
# Copyright (C) 2018-2024 Maxim Devaev <mdevaev@gmail.com> #
# #
# 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 <https://www.gnu.org/licenses/>. #
# #
*****************************************************************************/
#include "rtpc.h"
#include <linux/videodev2.h>
#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);
}

82
janus/src/rtpc.h Normal file
View File

@@ -0,0 +1,82 @@
/*****************************************************************************
# #
# uStreamer - Lightweight and fast MJPEG-HTTP streamer. #
# #
# Copyright (C) 2018-2024 Maxim Devaev <mdevaev@gmail.com> #
# #
# 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 <https://www.gnu.org/licenses/>. #
# #
*****************************************************************************/
#pragma once
#include <janus/rtp.h>
#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);

11
python/src/ustreamer.c
View File

@@ -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();

176
src/libs/memsink.c
View File

@@ -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;
}

28
src/libs/memsink.h
View File

@@ -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);

15
src/libs/memsinksh.c
View File

@@ -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()
);
}

12
src/libs/memsinksh.h
View File

@@ -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);