forked/ustreamer
1
0
Fork 0
mirror of https://github.com/pikvm/ustreamer.git synced 2026-03-03 14:21:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #33 from pikvm/sem-timeout

Browse Source 
Sem timeout
This commit is contained in:
Maxim Devaev
2020-08-19 13:44:23 +03:00
committed by GitHub
parent 4e51439118 191f6e3c09
commit 4faabf27ec
6 changed files with 35 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/device.c
View File

@@ -120,6 +120,7 @@ struct device_t *device_init(void) {
dev->standard = V4L2_STD_UNKNOWN;
dev->n_buffers = cores_available + 1;
dev->n_workers = min_u(cores_available, dev->n_buffers);
dev->min_frame_size = 128;
dev->timeout = 1;
dev->error_delay = 1;
dev->io_method = V4L2_MEMORY_MMAP;

4
src/encoder.c
View File

@@ -201,16 +201,20 @@ int encoder_compress_buffer(struct encoder_t *encoder, struct device_t *dev, uns
#pragma GCC diagnostic pop
assert(encoder->run->type != ENCODER_TYPE_UNKNOWN);
assert(dev->run->hw_buffers[buf_index].used > 0);
dev->run->pictures[buf_index]->encode_begin_ts = get_now_monotonic();
if (encoder->run->type == ENCODER_TYPE_CPU) {
LOG_VERBOSE("Compressing buffer %u using CPU", buf_index);
cpu_encoder_compress_buffer(dev, buf_index, encoder->run->quality);
} else if (encoder->run->type == ENCODER_TYPE_HW) {
LOG_VERBOSE("Compressing buffer %u using HW (just copying)", buf_index);
hw_encoder_compress_buffer(dev, buf_index);
}
# ifdef WITH_OMX
else if (encoder->run->type == ENCODER_TYPE_OMX) {
LOG_VERBOSE("Compressing buffer %u using OMX", buf_index);
if (omx_encoder_compress_buffer(encoder->run->omxs[worker_number], dev, buf_index) < 0) {
goto error;
}

23
src/encoders/omx/encoder.c
View File

@@ -102,11 +102,11 @@ struct omx_encoder_t *omx_encoder_init(void) {
LOG_INFO("Initializing OMX encoder ...");
if (vcos_semaphore_create(&omx->handler_lock, "handler_lock", 0) != VCOS_SUCCESS) {
if (vcos_semaphore_create(&omx->handler_sem, "handler_sem", 0) != VCOS_SUCCESS) {
LOG_ERROR("Can't create VCOS semaphore");
goto error;
}
omx->i_handler_lock = true;
omx->i_handler_sem = true;
if (_omx_init_component(omx) < 0) {
goto error;
@@ -132,8 +132,8 @@ void omx_encoder_destroy(struct omx_encoder_t *omx) {
_omx_encoder_clear_ports(omx);
component_set_state(&omx->encoder, OMX_StateLoaded);
if (omx->i_handler_lock) {
vcos_semaphore_delete(&omx->handler_lock);
if (omx->i_handler_sem) {
vcos_semaphore_delete(&omx->handler_sem);
}
if (omx->i_encoder) {
@@ -180,6 +180,7 @@ int omx_encoder_compress_buffer(struct omx_encoder_t *omx, struct device_t *dev,
# define OUT(_next) omx->output_buffer->_next
OMX_ERRORTYPE error;
VCOS_STATUS_T sem_status;
size_t slice_size = (IN(nAllocLen) < HW_BUFFER(used) ? IN(nAllocLen) : HW_BUFFER(used));
size_t pos = 0;
@@ -236,7 +237,13 @@ int omx_encoder_compress_buffer(struct omx_encoder_t *omx, struct device_t *dev,
}
}
vcos_semaphore_wait(&omx->handler_lock);
// vcos_semaphore_wait(&omx->handler_sem);
switch (sem_status = vcos_semaphore_wait_timeout(&omx->handler_sem, 3000)) {
case VCOS_SUCCESS: break;
case VCOS_EAGAIN: LOG_ERROR("Can't wait VCOS semaphore: EAGAIN (timeout)"); return -1;
case VCOS_EINVAL: LOG_ERROR("Can't wait VCOS semaphore: EINTVAL"); return -1;
default: LOG_ERROR("Can't wait VCOS semaphore: %d", sem_status); return -1;
}
}
# undef OUT
@@ -466,7 +473,7 @@ static OMX_ERRORTYPE _omx_event_handler(
if (event == OMX_EventError) {
LOG_ERROR_OMX((OMX_ERRORTYPE)data1, "OMX error event received");
omx->failed = true;
vcos_semaphore_post(&omx->handler_lock);
assert(vcos_semaphore_post(&omx->handler_sem) == VCOS_SUCCESS);
}
return OMX_ErrorNone;
}
@@ -481,7 +488,7 @@ static OMX_ERRORTYPE _omx_input_required_handler(
struct omx_encoder_t *omx = (struct omx_encoder_t *)v_omx;
omx->input_required = true;
vcos_semaphore_post(&omx->handler_lock);
assert(vcos_semaphore_post(&omx->handler_sem) == VCOS_SUCCESS);
return OMX_ErrorNone;
}
@@ -495,6 +502,6 @@ static OMX_ERRORTYPE _omx_output_available_handler(
struct omx_encoder_t *omx = (struct omx_encoder_t *)v_omx;
omx->output_available = true;
vcos_semaphore_post(&omx->handler_lock);
assert(vcos_semaphore_post(&omx->handler_sem) == VCOS_SUCCESS);
return OMX_ErrorNone;
}

4
src/encoders/omx/encoder.h
View File

@@ -42,9 +42,9 @@ struct omx_encoder_t {
bool input_required;
bool output_available;
bool failed;
VCOS_SEMAPHORE_T handler_lock;
VCOS_SEMAPHORE_T handler_sem;
bool i_handler_lock;
bool i_handler_sem;
bool i_encoder;
bool i_input_port_enabled;
bool i_output_port_enabled;

4
src/options.c
View File

@@ -339,7 +339,7 @@ int options_parse(struct options_t *options, struct device_t *dev, struct encode
case _O_TV_STANDARD: OPT_PARSE("TV standard", dev->standard, device_parse_standard, STANDARD_UNKNOWN, STANDARDS_STR);
case _O_IO_METHOD: OPT_PARSE("IO method", dev->io_method, device_parse_io_method, IO_METHOD_UNKNOWN, IO_METHODS_STR);
case _O_DESIRED_FPS: OPT_NUMBER("--desired-fps", dev->desired_fps, 0, VIDEO_MAX_FPS, 0);
case _O_MIN_FRAME_SIZE: OPT_NUMBER("--min-frame-size", dev->min_frame_size, 0, 8192, 0);
case _O_MIN_FRAME_SIZE: OPT_NUMBER("--min-frame-size", dev->min_frame_size, 1, 8192, 0);
case _O_PERSISTENT: OPT_SET(dev->persistent, true);
case _O_DV_TIMINGS: OPT_SET(dev->dv_timings, true);
case _O_BUFFERS: OPT_NUMBER("--buffers", dev->n_buffers, 1, 32, 0);
@@ -571,7 +571,7 @@ static void _help(struct device_t *dev, struct encoder_t *encoder, struct http_s
printf(" Available: %s; default: MMAP\n\n", IO_METHODS_STR);
printf(" -f|--desired-fps <N> ──────────────── Desired FPS. Default: maximum possible.\n\n");
printf(" -z|--min-frame-size <N> ───────────── Drop frames smaller then this limit. Useful if the device\n");
printf(" produces small-sized garbage frames. Default: disabled.\n\n");
printf(" produces small-sized garbage frames. Default: %zu bytes.\n\n", dev->min_frame_size);
printf(" -n|--persistent ───────────────────── Don't re-initialize device on timeout. Default: disabled.\n\n");
printf(" -t|--dv-timings ───────────────────── Enable DV timings querying and events processing\n");
printf(" to automatic resolution change. Default: disabled.\n\n");

21
src/stream.c
View File

@@ -449,23 +449,24 @@ static void *_worker_thread(void *v_worker) {
GPIO_SET_HIGH_AT(workers_busy_at, worker->number);
# endif
if (encoder_compress_buffer(worker->encoder, worker->dev, worker->number, worker->buf_index) < 0) {
worker->job_failed = false;
}
worker->job_failed = (bool)encoder_compress_buffer(worker->encoder, worker->dev, worker->number, worker->buf_index);
if (device_release_buffer(worker->dev, worker->buf_index) == 0) {
worker->job_start_ts = PICTURE(encode_begin_ts);
atomic_store(&worker->has_job, false);
if (!worker->job_failed) {
worker->job_start_ts = PICTURE(encode_begin_ts);
worker->last_comp_time = PICTURE(encode_end_ts) - worker->job_start_ts;
worker->last_comp_time = PICTURE(encode_end_ts) - worker->job_start_ts;
LOG_VERBOSE("Compressed new JPEG: size=%zu, time=%0.3Lf, worker=%u, buffer=%u",
PICTURE(used), worker->last_comp_time, worker->number, worker->buf_index);
LOG_VERBOSE("Compressed new JPEG: size=%zu, time=%0.3Lf, worker=%u, buffer=%u",
PICTURE(used), worker->last_comp_time, worker->number, worker->buf_index);
} else {
LOG_VERBOSE("Compression failed: worker=%u, buffer=%u", worker->number, worker->buf_index);
}
} else {
worker->job_failed = true;
atomic_store(&worker->has_job, false);
}
atomic_store(&worker->has_job, false);
# undef PICTURE
}