forked/ustreamer
1
0
Fork 0
mirror of https://github.com/pikvm/ustreamer.git synced 2026-02-18 02:55:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

parallel OMX encoding

Browse Source
This commit is contained in:
Devaev Maxim 2018-11-05 10:12:22 +03:00
parent da0e00252d
commit e5a57ac2e0
8 changed files with 88 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/device.c
View File

@ -138,8 +138,6 @@ int device_open(struct device_t *dev) {
}
_device_open_alloc_picbufs(dev);
dev->run->n_workers = dev->n_workers;
LOG_DEBUG("Device fd=%d initialized", dev->run->fd);
return 0;

60
src/encoder.c
View File

@ -54,7 +54,11 @@ struct encoder_t *encoder_init() {
return encoder;
}
void encoder_prepare(struct encoder_t *encoder) {
#pragma GCC diagnostic ignored "-Wunused-parameter"
#pragma GCC diagnostic push
void encoder_prepare(struct encoder_t *encoder, struct device_t *dev) {
#pragma GCC diagnostic pop
assert(encoder->type != ENCODER_TYPE_UNKNOWN);
if (encoder->type != ENCODER_TYPE_CPU) {
@ -65,8 +69,20 @@ void encoder_prepare(struct encoder_t *encoder) {
# ifdef OMX_ENCODER
if (encoder->type == ENCODER_TYPE_OMX) {
if ((encoder->omx = omx_encoder_init()) == NULL) {
goto use_fallback;
if (dev->n_workers > OMX_MAX_ENCODERS) {
LOG_INFO(
"OMX-based encoder can only work with %u worker threads; forced --workers=%u",
OMX_MAX_ENCODERS, OMX_MAX_ENCODERS
);
dev->n_workers = OMX_MAX_ENCODERS;
}
encoder->n_omxs = dev->n_workers;
A_CALLOC(encoder->omxs, encoder->n_omxs);
for (unsigned index = 0; index < encoder->n_omxs; ++index) {
if ((encoder->omxs[index] = omx_encoder_init()) == NULL) {
goto use_fallback;
}
}
}
# endif
@ -83,8 +99,13 @@ void encoder_prepare(struct encoder_t *encoder) {
void encoder_destroy(struct encoder_t *encoder) {
# ifdef OMX_ENCODER
if (encoder->omx) {
omx_encoder_destroy(encoder->omx);
if (encoder->omxs) {
for (unsigned index = 0; index < encoder->n_omxs; ++index) {
if (encoder->omxs[index]) {
omx_encoder_destroy(encoder->omxs[index]);
}
}
free(encoder->omxs);
}
# endif
free(encoder);
@ -101,18 +122,16 @@ enum encoder_type_t encoder_parse_type(const char *const str) {
#pragma GCC diagnostic ignored "-Wunused-parameter"
#pragma GCC diagnostic push
void encoder_prepare_for_device(struct encoder_t *encoder, struct device_t *dev) {
void encoder_prepare_live(struct encoder_t *encoder, struct device_t *dev) {
assert(encoder->type != ENCODER_TYPE_UNKNOWN);
#pragma GCC diagnostic pop
# ifdef OMX_ENCODER
if (encoder->type == ENCODER_TYPE_OMX) {
if (omx_encoder_prepare_for_device(encoder->omx, dev, encoder->quality, encoder->omx_use_ijg) < 0) {
goto use_fallback;
}
if (dev->run->n_workers > 1) {
LOG_INFO("OMX encoder can only work with one worker thread; forcing n_workers to 1");
dev->run->n_workers = 1;
for (unsigned index = 0; index < encoder->n_omxs; ++index) {
if (omx_encoder_prepare_live(encoder->omxs[index], dev, encoder->quality, encoder->omx_use_ijg) < 0) {
goto use_fallback;
}
}
}
# endif
@ -124,27 +143,31 @@ void encoder_prepare_for_device(struct encoder_t *encoder, struct device_t *dev)
use_fallback:
LOG_ERROR("Can't prepare selected encoder, falling back to CPU");
encoder->type = ENCODER_TYPE_CPU;
dev->run->n_workers = dev->n_workers;
# pragma GCC diagnostic pop
}
int encoder_compress_buffer(struct encoder_t *encoder, struct device_t *dev, const unsigned index) {
#pragma GCC diagnostic ignored "-Wunused-label"
#pragma GCC diagnostic push
int encoder_compress_buffer(struct encoder_t *encoder, struct device_t *dev,
const unsigned worker_number, const unsigned buf_index) {
#pragma GCC diagnostic pop
assert(encoder->type != ENCODER_TYPE_UNKNOWN);
dev->run->pictures[index].encode_begin_time = get_now_monotonic();
dev->run->pictures[buf_index].encode_begin_time = get_now_monotonic();
if (encoder->type == ENCODER_TYPE_CPU) {
jpeg_encoder_compress_buffer(dev, index, encoder->quality);
jpeg_encoder_compress_buffer(dev, buf_index, encoder->quality);
}
# ifdef OMX_ENCODER
else if (encoder->type == ENCODER_TYPE_OMX) {
if (omx_encoder_compress_buffer(encoder->omx, dev, index) < 0) {
if (omx_encoder_compress_buffer(encoder->omxs[worker_number], dev, buf_index) < 0) {
goto error;
}
}
# endif
dev->run->pictures[index].encode_end_time = get_now_monotonic();
dev->run->pictures[buf_index].encode_end_time = get_now_monotonic();
return 0;
@ -153,7 +176,6 @@ int encoder_compress_buffer(struct encoder_t *encoder, struct device_t *dev, con
error:
LOG_INFO("HW compressing error, falling back to CPU");
encoder->type = ENCODER_TYPE_CPU;
dev->run->n_workers = dev->n_workers;
return -1;
# pragma GCC diagnostic pop
}

11
src/encoder.h
View File

@ -49,7 +49,8 @@ struct encoder_t {
unsigned quality;
#ifdef OMX_ENCODER
bool omx_use_ijg;
struct omx_encoder_t *omx;
unsigned n_omxs;
struct omx_encoder_t **omxs;
#endif
};
@ -59,6 +60,8 @@ void encoder_destroy(struct encoder_t *encoder);
enum encoder_type_t encoder_parse_type(const char *const str);
void encoder_prepare(struct encoder_t *encoder);
void encoder_prepare_for_device(struct encoder_t *encoder, struct device_t *dev);
int encoder_compress_buffer(struct encoder_t *encoder, struct device_t *dev, const unsigned index);
void encoder_prepare(struct encoder_t *encoder, struct device_t *dev);
void encoder_prepare_live(struct encoder_t *encoder, struct device_t *dev);
int encoder_compress_buffer(struct encoder_t *encoder, struct device_t *dev,
const unsigned worker_number, const unsigned buf_index);

2
src/main.c
View File

@ -287,7 +287,7 @@ int main(int argc, char *argv[]) {
if ((exit_code = _parse_options(argc, argv, dev, encoder, server)) == 0) {
_install_signal_handlers();
encoder_prepare(encoder);
encoder_prepare(encoder, dev);
pthread_t stream_loop_tid;
pthread_t server_loop_tid;

36
src/omx/encoder.c
View File

@ -45,6 +45,9 @@
#define OUTPUT_PORT 341
static int _i_omx = 0;
static int _omx_init_component(struct omx_encoder_t *omx);
static int _omx_init_disable_ports(struct omx_encoder_t *omx);
static int _omx_setup_input(struct omx_encoder_t *omx, struct device_t *dev);
@ -77,17 +80,20 @@ struct omx_encoder_t *omx_encoder_init() {
A_CALLOC(omx, 1);
LOG_INFO("Initializing OMX JPEG encoder ...");
assert(_i_omx >= 0);
if (_i_omx == 0) {
LOG_INFO("Initializing BCM ...");
bcm_host_init();
LOG_DEBUG("Initializing BCM ...");
bcm_host_init();
LOG_DEBUG("Initializing OMX ...");
if ((error = OMX_Init()) != OMX_ErrorNone) {
LOG_OMX_ERROR(error, "Can't initialize OMX");
goto error;
LOG_INFO("Initializing OMX ...");
if ((error = OMX_Init()) != OMX_ErrorNone) {
LOG_OMX_ERROR(error, "Can't initialize OMX");
goto error;
}
}
omx->i_omx = true;
_i_omx += 1;
LOG_INFO("Initializing OMX JPEG encoder ...");
if (vcos_semaphore_create(&omx->handler_lock, "handler_lock", 0) != VCOS_SUCCESS) {
LOG_ERROR("Can't create VCOS semaphore");
@ -129,14 +135,20 @@ void omx_encoder_destroy(struct omx_encoder_t *omx) {
}
}
if (omx->i_omx) {
assert(_i_omx >= 0);
_i_omx -= 1;
if (_i_omx == 0) {
LOG_INFO("Destroying OMX ...");
OMX_Deinit();
LOG_INFO("Destroying BCM ...");
bcm_host_deinit();
}
bcm_host_deinit();
free(omx);
}
int omx_encoder_prepare_for_device(struct omx_encoder_t *omx, struct device_t *dev, const unsigned quality, const bool use_ijg) {
int omx_encoder_prepare_live(struct omx_encoder_t *omx, struct device_t *dev, const unsigned quality, const bool use_ijg) {
if (component_set_state(&omx->encoder, OMX_StateIdle) < 0) {
return -1;
}

6
src/omx/encoder.h
View File

@ -29,6 +29,9 @@
#include "../device.h"
#define OMX_MAX_ENCODERS 3
struct omx_encoder_t {
OMX_HANDLETYPE encoder;
OMX_BUFFERHEADERTYPE *input_buffer;
@ -38,7 +41,6 @@ struct omx_encoder_t {
bool failed;
VCOS_SEMAPHORE_T handler_lock;
bool i_omx;
bool i_handler_lock;
bool i_encoder;
bool i_input_port_enabled;
@ -49,5 +51,5 @@ struct omx_encoder_t {
struct omx_encoder_t *omx_encoder_init();
void omx_encoder_destroy(struct omx_encoder_t *omx);
int omx_encoder_prepare_for_device(struct omx_encoder_t *omx, struct device_t *dev, const unsigned quality, const bool use_ijg);
int omx_encoder_prepare_live(struct omx_encoder_t *omx, struct device_t *dev, const unsigned quality, const bool use_ijg);
int omx_encoder_compress_buffer(struct omx_encoder_t *omx, struct device_t *dev, const unsigned index);

20
src/stream.c
View File

@ -116,7 +116,7 @@ void stream_loop(struct stream_t *stream) {
LOG_PERF("##### Raw frame accepted; worker = %u", free_worker_number);
} else {
for (unsigned number = 0; number < stream->dev->run->n_workers; ++number) {
for (unsigned number = 0; number < stream->dev->n_workers; ++number) {
if (!pool.workers[number].has_job && (free_worker_number == -1
|| pool.workers[free_worker_number].job_start_time < pool.workers[number].job_start_time
)) {
@ -325,16 +325,16 @@ static long double _stream_get_fluency_delay(struct device_t *dev, struct worker
long double min_delay;
long double soft_delay;
for (unsigned number = 0; number < dev->run->n_workers; ++number) {
for (unsigned number = 0; number < dev->n_workers; ++number) {
A_PTHREAD_M_LOCK(&pool->workers[number].last_comp_time_mutex);
if (pool->workers[number].last_comp_time > 0) {
comp_time += pool->workers[number].last_comp_time;
}
A_PTHREAD_M_UNLOCK(&pool->workers[number].last_comp_time_mutex);
}
comp_time = comp_time / dev->run->n_workers; // Среднее время работы воркеров
comp_time = comp_time / dev->n_workers; // Среднее время работы воркеров
min_delay = comp_time / dev->run->n_workers; // Минимальное время работы размазывается на N воркеров
min_delay = comp_time / dev->n_workers; // Минимальное время работы размазывается на N воркеров
soft_delay = ((long double)1) / dev->soft_fps; // Искусственное время задержки на основе желаемого FPS
if (min_delay > 0) {
@ -372,7 +372,7 @@ static int _stream_init(struct device_t *dev, struct workers_pool_t *pool) {
goto error;
}
encoder_prepare_for_device(pool->encoder, dev);
encoder_prepare_live(pool->encoder, dev);
_stream_init_workers(dev, pool);
@ -384,15 +384,15 @@ static int _stream_init(struct device_t *dev, struct workers_pool_t *pool) {
}
static void _stream_init_workers(struct device_t *dev, struct workers_pool_t *pool) {
LOG_INFO("Spawning %d workers ...", dev->run->n_workers);
LOG_INFO("Spawning %d workers ...", dev->n_workers);
*pool->workers_stop = false;
A_CALLOC(pool->workers, dev->run->n_workers);
A_CALLOC(pool->workers, dev->n_workers);
A_PTHREAD_M_INIT(&pool->free_workers_mutex);
A_PTHREAD_C_INIT(&pool->free_workers_cond);
for (unsigned number = 0; number < dev->run->n_workers; ++number) {
for (unsigned number = 0; number < dev->n_workers; ++number) {
pool->free_workers += 1;
A_PTHREAD_M_INIT(&pool->workers[number].has_job_mutex);
@ -440,7 +440,7 @@ static void *_stream_worker_thread(void *v_ctx) {
if (!*ctx->workers_stop) {
LOG_DEBUG("Worker %u compressing JPEG from buffer %d ...", ctx->number, ctx->buf_index);
if (encoder_compress_buffer(ctx->encoder, ctx->dev, ctx->buf_index) < 0) {
if (encoder_compress_buffer(ctx->encoder, ctx->dev, ctx->number, ctx->buf_index) < 0) {
*ctx->job_failed = true;
}
@ -479,7 +479,7 @@ static void _stream_destroy_workers(struct device_t *dev, struct workers_pool_t
LOG_INFO("Destroying workers ...");
*pool->workers_stop = true;
for (unsigned number = 0; number < dev->run->n_workers; ++number) {
for (unsigned number = 0; number < dev->n_workers; ++number) {
A_PTHREAD_M_LOCK(&pool->workers[number].has_job_mutex);
pool->workers[number].has_job = true; // Final job: die
A_PTHREAD_M_UNLOCK(&pool->workers[number].has_job_mutex);

2
src/stream.h
View File

@ -72,7 +72,7 @@ struct worker_t {
};
struct workers_pool_t {
struct worker_t*workers;
struct worker_t *workers;
bool *workers_stop;
pthread_mutex_t free_workers_mutex;