forked/ustreamer
1
0
Fork 0
mirror of https://github.com/pikvm/ustreamer.git synced 2025-12-23 18:50:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

wait (select) device in grab function

Browse Source
This commit is contained in:
Maxim Devaev 2024-03-02 09:33:45 +02:00
parent 85308e48fd
commit a00f49331c
4 changed files with 137 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

160
src/libs/device.c
View File

@ -80,6 +80,7 @@ static const struct {
{"USERPTR", V4L2_MEMORY_USERPTR},
};
static int _device_wait_buffer(us_device_s *dev);
static int _device_consume_event(us_device_s *dev);
static void _v4l2_buffer_copy(const struct v4l2_buffer *src, struct v4l2_buffer *dest);
static bool _device_is_buffer_valid(us_device_s *dev, const struct v4l2_buffer *buf, const u8 *data);
@ -258,77 +259,23 @@ void us_device_close(us_device_s *dev) {
run->persistent_timeout_reported = false;
}
int us_device_wait_buffer(us_device_s *dev) {
us_device_runtime_s *const run = dev->run;
# define INIT_FD_SET(x_set) \
fd_set x_set; FD_ZERO(&x_set); FD_SET(run->fd, &x_set);
INIT_FD_SET(read_fds);
INIT_FD_SET(error_fds);
# undef INIT_FD_SET
// Раньше мы проверяли и has_write, но потом выяснилось, что libcamerify зачем-то
// генерирует эвенты на запись, вероятно ошибочно. Судя по всему, игнорирование
// has_write не делает никому плохо.
struct timeval timeout;
timeout.tv_sec = dev->timeout;
timeout.tv_usec = 0;
_D_LOG_DEBUG("Calling select() on video device ...");
bool has_read = false;
bool has_error = false;
const int selected = select(run->fd + 1, &read_fds, NULL, &error_fds, &timeout);
if (selected > 0) {
has_read = FD_ISSET(run->fd, &read_fds);
has_error = FD_ISSET(run->fd, &error_fds);
}
_D_LOG_DEBUG("Device select() --> %d; has_read=%d, has_error=%d", selected, has_read, has_error);
if (selected < 0) {
if (errno != EINTR) {
_D_LOG_PERROR("Device select() error");
}
return -1;
} else if (selected == 0) {
if (dev->persistent) {
if (!run->persistent_timeout_reported) {
_D_LOG_ERROR("Persistent device timeout (unplugged)");
run->persistent_timeout_reported = true;
}
} else {
// Если устройство не персистентное, то таймаут является ошибкой
return -1;
}
return -2; // No new frames
} else {
run->persistent_timeout_reported = false;
if (has_error && _device_consume_event(dev) < 0) {
return -1; // Restart required
}
}
return 0;
}
static int _device_consume_event(us_device_s *dev) {
struct v4l2_event event;
if (us_xioctl(dev->run->fd, VIDIOC_DQEVENT, &event) < 0) {
_D_LOG_PERROR("Can't consume V4L2 event");
return -1;
}
switch (event.type) {
case V4L2_EVENT_SOURCE_CHANGE:
_D_LOG_INFO("Got V4L2_EVENT_SOURCE_CHANGE: Source changed");
return -1;
case V4L2_EVENT_EOS:
_D_LOG_INFO("Got V4L2_EVENT_EOS: End of stream");
return -1;
}
return 0;
}
int us_device_grab_buffer(us_device_s *dev, us_hw_buffer_s **hw) {
// Это сложная функция, которая делает сразу много всего, чтобы получить новый фрейм.
// - Вызывается _device_wait_buffer() с select() внутри, чтобы подождать новый фрейм
// или эвент V4L2. Обработка эвентов более приоритетна, чем кадров.
// - При таймауте select() в _device_wait_buffer() возвращаем -2 для персистентных
// устройств типа TC358743, для остальных же возвращаем ошибку -1.
// - Если есть новые фреймы, то пропустить их все, пока не закончатся и вернуть
// самый-самый свежий, содержащий при этом валидные данные.
// - Если таковых не нашлось, вернуть -3.
// - Ошибка -1 возвращается при любых сбоях.
switch (_device_wait_buffer(dev)) {
case 0: break; // New frame
case -2: return -2; // Persistent timeout
default: return -1; // Error
}
us_device_runtime_s *const run = dev->run;
*hw = NULL;
@ -409,7 +356,7 @@ int us_device_grab_buffer(us_device_s *dev, us_hw_buffer_s **hw) {
if (buf_got) {
break; // Process any latest valid frame
} else if (broken) {
return -2; // If we have only broken frames on this capture session
return -3; // If we have only broken frames on this capture session
}
}
_D_LOG_PERROR("Can't grab device buffer");
@ -444,6 +391,75 @@ int us_device_release_buffer(us_device_s *dev, us_hw_buffer_s *hw) {
return 0;
}
int _device_wait_buffer(us_device_s *dev) {
us_device_runtime_s *const run = dev->run;
# define INIT_FD_SET(x_set) \
fd_set x_set; FD_ZERO(&x_set); FD_SET(run->fd, &x_set);
INIT_FD_SET(read_fds);
INIT_FD_SET(error_fds);
# undef INIT_FD_SET
// Раньше мы проверяли и has_write, но потом выяснилось, что libcamerify зачем-то
// генерирует эвенты на запись, вероятно ошибочно. Судя по всему, игнорирование
// has_write не делает никому плохо.
struct timeval timeout;
timeout.tv_sec = dev->timeout;
timeout.tv_usec = 0;
_D_LOG_DEBUG("Calling select() on video device ...");
bool has_read = false;
bool has_error = false;
const int selected = select(run->fd + 1, &read_fds, NULL, &error_fds, &timeout);
if (selected > 0) {
has_read = FD_ISSET(run->fd, &read_fds);
has_error = FD_ISSET(run->fd, &error_fds);
}
_D_LOG_DEBUG("Device select() --> %d; has_read=%d, has_error=%d", selected, has_read, has_error);
if (selected < 0) {
if (errno != EINTR) {
_D_LOG_PERROR("Device select() error");
}
return -1;
} else if (selected == 0) {
if (!dev->persistent) {
// Если устройство не персистентное, то таймаут является ошибкой
return -1;
}
if (!run->persistent_timeout_reported) {
_D_LOG_ERROR("Persistent device timeout (unplugged)");
run->persistent_timeout_reported = true;
}
return -2; // Таймаут, нет новых фреймов
} else {
run->persistent_timeout_reported = false;
if (has_error && _device_consume_event(dev) < 0) {
return -1; // Restart required
}
}
return 0;
}
static int _device_consume_event(us_device_s *dev) {
struct v4l2_event event;
if (us_xioctl(dev->run->fd, VIDIOC_DQEVENT, &event) < 0) {
_D_LOG_PERROR("Can't consume V4L2 event");
return -1;
}
switch (event.type) {
case V4L2_EVENT_SOURCE_CHANGE:
_D_LOG_INFO("Got V4L2_EVENT_SOURCE_CHANGE: Source changed");
return -1;
case V4L2_EVENT_EOS:
_D_LOG_INFO("Got V4L2_EVENT_EOS: End of stream");
return -1;
}
return 0;
}
static void _v4l2_buffer_copy(const struct v4l2_buffer *src, struct v4l2_buffer *dest) {
struct v4l2_plane *dest_planes = dest->m.planes;
memcpy(dest, src, sizeof(struct v4l2_buffer));
@ -666,7 +682,7 @@ static int _device_open_format(us_device_s *dev, bool first) {
_format_to_string_supported(FMT(pixelformat)));
char *format_str;
if ((format_str = (char *)_format_to_string_nullable(FMT(pixelformat))) != NULL) {
if ((format_str = (char*)_format_to_string_nullable(FMT(pixelformat))) != NULL) {
_D_LOG_INFO("Falling back to format=%s", format_str);
} else {
char fourcc_str[8];

1
src/libs/device.h
View File

@ -132,6 +132,5 @@ int us_device_parse_io_method(const char *str);
int us_device_open(us_device_s *dev);
void us_device_close(us_device_s *dev);
int us_device_wait_buffer(us_device_s *dev);
int us_device_grab_buffer(us_device_s *dev, us_hw_buffer_s **hw);
int us_device_release_buffer(us_device_s *dev, us_hw_buffer_s *hw);

80
src/ustreamer/stream.c
View File

@ -98,7 +98,7 @@ void us_stream_loop(us_stream_s *stream) {
US_LOG_DEBUG("Waiting for worker ...");
us_worker_s *const ready_wr = us_workers_pool_wait(stream->enc->run->pool);
us_encoder_job_s *const ready_job = (us_encoder_job_s *)(ready_wr->job);
us_encoder_job_s *const ready_job = ready_wr->job;
if (ready_job->hw != NULL) {
if (us_device_release_buffer(stream->dev, ready_job->hw) < 0) {
@ -107,7 +107,7 @@ void us_stream_loop(us_stream_s *stream) {
ready_job->hw = NULL;
if (ready_wr->job_failed) {
// pass
} else if (ready_wr->job_timely) {
} else if (ready_wr->job_timely) {
_stream_expose_frame(stream, ready_job->dest);
US_LOG_PERF("##### Encoded JPEG exposed; worker=%s, latency=%.3Lf",
ready_wr->name, us_get_now_monotonic() - ready_job->dest->grab_ts);
@ -121,56 +121,50 @@ void us_stream_loop(us_stream_s *stream) {
goto close;
}
switch (us_device_wait_buffer(stream->dev)) {
case 0: break; // New frame
us_hw_buffer_s *hw;
const int buf_index = us_device_grab_buffer(stream->dev, &hw);
switch (buf_index) {
case -3: continue; // Broken frame
case -2: // Persistent timeout
# ifdef WITH_GPIO
us_gpio_set_stream_online(false);
# endif
goto close;
default: goto close; // Error
case -1: goto close; // Any error
}
assert(buf_index >= 0);
# ifdef WITH_GPIO
us_gpio_set_stream_online(true);
# endif
const long double now = us_get_now_monotonic();
const long long now_second = us_floor_ms(now);
us_hw_buffer_s *hw;
const int buf_index = us_device_grab_buffer(stream->dev, &hw);
if (buf_index >= 0) {
if (now < grab_after) {
fluency_passed += 1;
US_LOG_VERBOSE("Passed %u frames for fluency: now=%.03Lf, grab_after=%.03Lf",
fluency_passed, now, grab_after);
if (us_device_release_buffer(stream->dev, hw) < 0) {
goto close;
}
} else {
fluency_passed = 0;
if (now_second != captured_fps_second) {
US_LOG_PERF_FPS("A new second has come; captured_fps=%u", captured_fps_accum);
atomic_store(&run->captured_fps, captured_fps_accum);
captured_fps_accum = 0;
captured_fps_second = now_second;
}
captured_fps_accum += 1;
const long double fluency_delay = us_workers_pool_get_fluency_delay(stream->enc->run->pool, ready_wr);
grab_after = now + fluency_delay;
US_LOG_VERBOSE("Fluency: delay=%.03Lf, grab_after=%.03Lf", fluency_delay, grab_after);
ready_job->hw = hw;
us_workers_pool_assign(stream->enc->run->pool, ready_wr);
US_LOG_DEBUG("Assigned new frame in buffer=%d to worker=%s", buf_index, ready_wr->name);
_SINK_PUT(raw_sink, &hw->raw);
_H264_PUT(&hw->raw, h264_force_key);
if (now < grab_after) {
fluency_passed += 1;
US_LOG_VERBOSE("Passed %u frames for fluency: now=%.03Lf, grab_after=%.03Lf",
fluency_passed, now, grab_after);
if (us_device_release_buffer(stream->dev, hw) < 0) {
goto close;
}
} else {
fluency_passed = 0;
const long long now_second = us_floor_ms(now);
if (now_second != captured_fps_second) {
US_LOG_PERF_FPS("A new second has come; captured_fps=%u", captured_fps_accum);
atomic_store(&run->captured_fps, captured_fps_accum);
captured_fps_accum = 0;
captured_fps_second = now_second;
}
captured_fps_accum += 1;
const long double fluency_delay = us_workers_pool_get_fluency_delay(stream->enc->run->pool, ready_wr);
grab_after = now + fluency_delay;
US_LOG_VERBOSE("Fluency: delay=%.03Lf, grab_after=%.03Lf", fluency_delay, grab_after);
ready_job->hw = hw;
us_workers_pool_assign(stream->enc->run->pool, ready_wr);
US_LOG_DEBUG("Assigned new frame in buffer=%d to worker=%s", buf_index, ready_wr->name);
_SINK_PUT(raw_sink, &hw->raw);
_H264_PUT(&hw->raw, h264_force_key);
}
}
@ -212,7 +206,7 @@ static bool _stream_is_stopped(us_stream_s *stream) {
}
static bool _stream_has_any_clients(us_stream_s *stream) {
us_stream_runtime_s *const run = stream->run;
const us_stream_runtime_s *const run = stream->run;
return (
atomic_load(&run->http_has_clients)
// has_clients синков НЕ обновляются в реальном времени

27
src/v4p/main.c
View File

@ -216,30 +216,27 @@ static void _main_loop(void) {
continue;
}
switch (us_device_wait_buffer(dev)) {
case 0: break; // New frame
us_hw_buffer_s *hw;
const int buf_index = us_device_grab_buffer(dev, &hw);
switch (buf_index) {
case -3: continue; // Broken frame
case -2: // Persistent timeout
if (us_drm_expose(drm, US_DRM_EXPOSE_NO_SIGNAL, NULL, 0) < 0) {
_slowdown();
continue;
}
default: goto close; // Error
case -1: goto close; // Any error
}
assert(buf_index >= 0);
us_hw_buffer_s *hw;
const int buf_index = us_device_grab_buffer(dev, &hw);
if (buf_index >= 0) {
const int exposed = us_drm_expose(drm, US_DRM_EXPOSE_FRAME, &hw->raw, dev->run->hz);
if (us_device_release_buffer(dev, hw) < 0) {
goto close;
}
if (exposed < 0) {
_slowdown();
continue;
}
} else if (buf_index != -2) { // -2 for broken frame
const int exposed = us_drm_expose(drm, US_DRM_EXPOSE_FRAME, &hw->raw, dev->run->hz);
if (us_device_release_buffer(dev, hw) < 0) {
goto close;
}
if (exposed < 0) {
_slowdown();
continue;
}
}
close: