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

refactoring

Browse Source
This commit is contained in:
Devaev Maxim 2018-09-18 22:53:44 +03:00
parent 0d86ae146c
commit eb1b61a144
6 changed files with 52 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/device.c
View File

@ -53,12 +53,9 @@ struct device_t *device_init() {
struct device_t *dev; struct device_t *dev;
A_CALLOC(run, 1, sizeof(*run)); A_CALLOC(run, 1, sizeof(*run));
MEMSET_ZERO_PTR(run);
run->fd = -1; run->fd = -1;
A_CALLOC(dev, 1, sizeof(*dev)); A_CALLOC(dev, 1, sizeof(*dev));
MEMSET_ZERO_PTR(dev);
dev->path = (char *)DEFAULT_DEVICE; dev->path = (char *)DEFAULT_DEVICE;
dev->width = 640; dev->width = 640;
dev->height = 480; dev->height = 480;

8
src/http.c
View File

@ -29,19 +29,17 @@ static void _http_add_header(struct evhttp_request *request, const char *key, co
struct http_server_t *http_server_init(struct stream_t *stream) { struct http_server_t *http_server_init(struct stream_t *stream) {
struct stream_t *exposed;
struct http_server_runtime_t *run; struct http_server_runtime_t *run;
struct http_server_t *server; struct http_server_t *server;
struct exposed_t *exposed;
exposed = stream_init(); A_CALLOC(exposed, 1, sizeof(*exposed));
A_CALLOC(run, 1, sizeof(*run)); A_CALLOC(run, 1, sizeof(*run));
MEMSET_ZERO_PTR(run);
run->stream = stream; run->stream = stream;
run->exposed = exposed; run->exposed = exposed;
A_CALLOC(server, 1, sizeof(*server)); A_CALLOC(server, 1, sizeof(*server));
MEMSET_ZERO_PTR(server);
server->host = (char *)DEFAULT_HOST; server->host = (char *)DEFAULT_HOST;
server->port = 8080; server->port = 8080;
server->run = run; server->run = run;
@ -61,7 +59,7 @@ void http_server_destroy(struct http_server_t *server) {
evhttp_free(server->run->http); evhttp_free(server->run->http);
event_base_free(server->run->base); event_base_free(server->run->base);
free(server->run->exposed->picture.data); free(server->run->exposed->picture.data);
stream_destroy(server->run->exposed); free(server->run->exposed);
free(server->run); free(server->run);
free(server); free(server);
libevent_global_shutdown(); libevent_global_shutdown();

9
src/http.h
View File

@ -6,11 +6,18 @@
#include "stream.h" #include "stream.h"
struct exposed_t {
struct picture_t picture;
unsigned width;
unsigned height;
bool online;
};
struct http_server_runtime_t { struct http_server_runtime_t {
struct event_base *base; struct event_base *base;
struct evhttp *http; struct evhttp *http;
struct stream_t *stream; struct stream_t *stream;
struct stream_t *exposed; // updated and mutex are not used struct exposed_t *exposed;
}; };
struct http_server_t { struct http_server_t {

8
src/main.c
View File

@ -104,7 +104,6 @@ static int _parse_options(int argc, char *argv[], struct device_t *dev, struct h
} }
struct main_context_t { struct main_context_t {
struct device_t *dev;
struct stream_t *stream; struct stream_t *stream;
struct http_server_t *server; struct http_server_t *server;
}; };
@ -121,7 +120,7 @@ static void _block_thread_signals() {
static void *_stream_loop_thread(UNUSED void *_) { static void *_stream_loop_thread(UNUSED void *_) {
_block_thread_signals(); _block_thread_signals();
stream_loop(_ctx->dev, _ctx->stream); stream_loop(_ctx->stream);
return NULL; return NULL;
} }
@ -133,7 +132,7 @@ static void *_server_loop_thread(UNUSED void *_) {
static void _signal_handler(int signum) { static void _signal_handler(int signum) {
LOG_INFO("===== Stopping by %s =====", strsignal(signum)); LOG_INFO("===== Stopping by %s =====", strsignal(signum));
stream_loop_break(_ctx->dev); stream_loop_break(_ctx->stream);
http_server_loop_break(_ctx->server); http_server_loop_break(_ctx->server);
} }
@ -163,7 +162,7 @@ int main(int argc, char *argv[]) {
int exit_code = 0; int exit_code = 0;
dev = device_init(); dev = device_init();
stream = stream_init(); stream = stream_init(dev);
server = http_server_init(stream); server = http_server_init(stream);
if ((exit_code = _parse_options(argc, argv, dev, server)) == 0) { if ((exit_code = _parse_options(argc, argv, dev, server)) == 0) {
@ -173,7 +172,6 @@ int main(int argc, char *argv[]) {
pthread_t server_loop_tid; pthread_t server_loop_tid;
struct main_context_t ctx; struct main_context_t ctx;
ctx.dev = dev;
ctx.stream = stream; ctx.stream = stream;
ctx.server = server; ctx.server = server;
_ctx = &ctx; _ctx = &ctx;

69
src/stream.c
View File

@ -32,12 +32,11 @@ static int _stream_release_buffer(struct device_t *dev, struct v4l2_buffer *buf_
static int _stream_handle_event(struct device_t *dev); static int _stream_handle_event(struct device_t *dev);
struct stream_t *stream_init() { struct stream_t *stream_init(struct device_t *dev) {
struct stream_t *stream; struct stream_t *stream;
A_CALLOC(stream, 1, sizeof(*stream)); A_CALLOC(stream, 1, sizeof(*stream));
MEMSET_ZERO_PTR(stream); stream->dev = dev;
A_PTHREAD_M_INIT(&stream->mutex); A_PTHREAD_M_INIT(&stream->mutex);
return stream; return stream;
} }
@ -68,17 +67,17 @@ static void _stream_dump(struct stream_t *stream) {
} }
#endif #endif
void stream_loop(struct device_t *dev, struct stream_t *stream) { void stream_loop(struct stream_t *stream) {
struct workers_pool_t pool; struct workers_pool_t pool;
bool workers_stop; bool workers_stop;
MEMSET_ZERO(pool); MEMSET_ZERO(pool);
pool.workers_stop = &workers_stop; pool.workers_stop = &workers_stop;
LOG_INFO("Using V4L2 device: %s", dev->path); LOG_INFO("Using V4L2 device: %s", stream->dev->path);
LOG_INFO("Using JPEG quality: %d%%", dev->jpeg_quality); LOG_INFO("Using JPEG quality: %d%%", stream->dev->jpeg_quality);
while (_stream_init_loop(dev, &pool) == 0) { while (_stream_init_loop(stream->dev, &pool) == 0) {
struct worker_t *last_worker = NULL; struct worker_t *last_worker = NULL;
unsigned frames_count = 0; unsigned frames_count = 0;
long double grab_after = 0; long double grab_after = 0;
@ -87,15 +86,15 @@ void stream_loop(struct device_t *dev, struct stream_t *stream) {
long long fps_second = 0; long long fps_second = 0;
LOG_DEBUG("Allocation memory for stream picture ..."); LOG_DEBUG("Allocation memory for stream picture ...");
A_CALLOC(stream->picture.data, dev->run->max_picture_size, sizeof(*stream->picture.data)); A_CALLOC(stream->picture.data, stream->dev->run->max_picture_size, sizeof(*stream->picture.data));
A_PTHREAD_M_LOCK(&stream->mutex); A_PTHREAD_M_LOCK(&stream->mutex);
stream->width = dev->run->width; stream->width = stream->dev->run->width;
stream->height = dev->run->height; stream->height = stream->dev->run->height;
stream->online = true; stream->online = true;
A_PTHREAD_M_UNLOCK(&stream->mutex); A_PTHREAD_M_UNLOCK(&stream->mutex);
while (!dev->stop) { while (!stream->dev->stop) {
SEP_DEBUG('-'); SEP_DEBUG('-');
LOG_DEBUG("Waiting for workers ..."); LOG_DEBUG("Waiting for workers ...");
@ -103,13 +102,13 @@ void stream_loop(struct device_t *dev, struct stream_t *stream) {
A_PTHREAD_C_WAIT_TRUE(pool.has_free_workers, &pool.has_free_workers_cond, &pool.has_free_workers_mutex); A_PTHREAD_C_WAIT_TRUE(pool.has_free_workers, &pool.has_free_workers_cond, &pool.has_free_workers_mutex);
A_PTHREAD_M_UNLOCK(&pool.has_free_workers_mutex); A_PTHREAD_M_UNLOCK(&pool.has_free_workers_mutex);
if (last_worker && !last_worker->has_job && dev->run->pictures[last_worker->ctx.index].data) { if (last_worker && !last_worker->has_job && stream->dev->run->pictures[last_worker->ctx.index].data) {
A_PTHREAD_M_LOCK(&stream->mutex); A_PTHREAD_M_LOCK(&stream->mutex);
stream->picture.size = dev->run->pictures[last_worker->ctx.index].size; stream->picture.size = stream->dev->run->pictures[last_worker->ctx.index].size;
stream->picture.allocated = dev->run->pictures[last_worker->ctx.index].allocated; stream->picture.allocated = stream->dev->run->pictures[last_worker->ctx.index].allocated;
memcpy( memcpy(
stream->picture.data, stream->picture.data,
dev->run->pictures[last_worker->ctx.index].data, stream->dev->run->pictures[last_worker->ctx.index].data,
stream->picture.size * sizeof(*stream->picture.data) stream->picture.size * sizeof(*stream->picture.data)
); );
stream->updated = true; stream->updated = true;
@ -122,23 +121,23 @@ void stream_loop(struct device_t *dev, struct stream_t *stream) {
# endif # endif
} }
if (dev->stop) { if (stream->dev->stop) {
break; break;
} }
# define INIT_FD_SET(_set) \ # define INIT_FD_SET(_set) \
fd_set _set; FD_ZERO(&_set); FD_SET(dev->run->fd, &_set); fd_set _set; FD_ZERO(&_set); FD_SET(stream->dev->run->fd, &_set);
INIT_FD_SET(read_fds); INIT_FD_SET(read_fds);
INIT_FD_SET(write_fds); INIT_FD_SET(write_fds);
INIT_FD_SET(error_fds); INIT_FD_SET(error_fds);
# undef INIT_FD_SET # undef INIT_FD_SET
struct timeval timeout; struct timeval timeout;
timeout.tv_sec = dev->timeout; timeout.tv_sec = stream->dev->timeout;
timeout.tv_usec = 0; timeout.tv_usec = 0;
LOG_DEBUG("Calling select() on video device ..."); LOG_DEBUG("Calling select() on video device ...");
int retval = select(dev->run->fd + 1, &read_fds, &write_fds, &error_fds, &timeout); int retval = select(stream->dev->run->fd + 1, &read_fds, &write_fds, &error_fds, &timeout);
LOG_DEBUG("Device select() --> %d", retval); LOG_DEBUG("Device select() --> %d", retval);
if (retval < 0) { if (retval < 0) {
@ -152,18 +151,18 @@ void stream_loop(struct device_t *dev, struct stream_t *stream) {
break; break;
} else { } else {
if (FD_ISSET(dev->run->fd, &read_fds)) { if (FD_ISSET(stream->dev->run->fd, &read_fds)) {
LOG_DEBUG("Frame is ready"); LOG_DEBUG("Frame is ready");
struct v4l2_buffer buf_info; struct v4l2_buffer buf_info;
if (_stream_grab_buffer(dev, &buf_info) < 0) { if (_stream_grab_buffer(stream->dev, &buf_info) < 0) {
break; break;
} }
if (dev->every_frame) { if (stream->dev->every_frame) {
if (frames_count < dev->every_frame - 1) { if (frames_count < stream->dev->every_frame - 1) {
LOG_DEBUG("Dropping frame %d for option --every-frame=%d", frames_count + 1, dev->every_frame); LOG_DEBUG("Dropping frame %d for option --every-frame=%d", frames_count + 1, stream->dev->every_frame);
++frames_count; ++frames_count;
goto pass_frame; goto pass_frame;
} }
@ -175,7 +174,7 @@ void stream_loop(struct device_t *dev, struct stream_t *stream) {
// The good thing is such frames are quite small compared to the regular pictures. // The good thing is such frames are quite small compared to the regular pictures.
// For example a VGA (640x480) webcam picture is normally >= 8kByte large, // For example a VGA (640x480) webcam picture is normally >= 8kByte large,
// corrupted frames are smaller. // corrupted frames are smaller.
if (buf_info.bytesused < dev->min_frame_size) { if (buf_info.bytesused < stream->dev->min_frame_size) {
LOG_DEBUG("Dropping too small frame sized %d bytes, assuming it as broken", buf_info.bytesused); LOG_DEBUG("Dropping too small frame sized %d bytes, assuming it as broken", buf_info.bytesused);
goto pass_frame; goto pass_frame;
} }
@ -199,7 +198,7 @@ void stream_loop(struct device_t *dev, struct stream_t *stream) {
++fps; ++fps;
} }
long double delay = _stream_get_fluency_delay(dev, &pool); long double delay = _stream_get_fluency_delay(stream->dev, &pool);
grab_after = now + delay; grab_after = now + delay;
LOG_PERF("Fluency delay=%.03Lf; grab_after=%.03Lf", delay, grab_after); LOG_PERF("Fluency delay=%.03Lf; grab_after=%.03Lf", delay, grab_after);
} }
@ -222,21 +221,21 @@ void stream_loop(struct device_t *dev, struct stream_t *stream) {
pass_frame: pass_frame:
if (_stream_release_buffer(dev, &buf_info) < 0) { if (_stream_release_buffer(stream->dev, &buf_info) < 0) {
break; break;
} }
} }
next_handlers: next_handlers:
if (FD_ISSET(dev->run->fd, &write_fds)) { if (FD_ISSET(stream->dev->run->fd, &write_fds)) {
LOG_ERROR("Got unexpected writing event, seems device was disconnected"); LOG_ERROR("Got unexpected writing event, seems device was disconnected");
break; break;
} }
if (FD_ISSET(dev->run->fd, &error_fds)) { if (FD_ISSET(stream->dev->run->fd, &error_fds)) {
LOG_INFO("Got V4L2 event"); LOG_INFO("Got V4L2 event");
if (_stream_handle_event(dev) < 0) { if (_stream_handle_event(stream->dev) < 0) {
break; break;
} }
} }
@ -250,13 +249,13 @@ void stream_loop(struct device_t *dev, struct stream_t *stream) {
A_PTHREAD_M_UNLOCK(&stream->mutex); A_PTHREAD_M_UNLOCK(&stream->mutex);
} }
_stream_destroy_workers(dev, &pool); _stream_destroy_workers(stream->dev, &pool);
_stream_control(dev, false); _stream_control(stream->dev, false);
device_close(dev); device_close(stream->dev);
} }
void stream_loop_break(struct device_t *dev) { void stream_loop_break(struct stream_t *stream) {
dev->stop = 1; stream->dev->stop = 1;
} }
static long double _stream_get_fluency_delay(struct device_t *dev, struct workers_pool_t *pool) { static long double _stream_get_fluency_delay(struct device_t *dev, struct workers_pool_t *pool) {

9
src/stream.h
View File

@ -1,10 +1,8 @@
#pragma once #pragma once
#include <signal.h>
#include <stdbool.h> #include <stdbool.h>
#include <signal.h> #include <signal.h>
#include <pthread.h> #include <pthread.h>
#include <linux/videodev2.h>
#include "device.h" #include "device.h"
@ -58,11 +56,12 @@ struct stream_t {
bool online; bool online;
bool updated; bool updated;
pthread_mutex_t mutex; pthread_mutex_t mutex;
struct device_t *dev;
}; };
struct stream_t *stream_init(); struct stream_t *stream_init(struct device_t *dev);
void stream_destroy(struct stream_t *stream); void stream_destroy(struct stream_t *stream);
void stream_loop(struct device_t *dev, struct stream_t *stream); void stream_loop(struct stream_t *stream);
void stream_loop_break(struct device_t *dev); void stream_loop_break(struct stream_t *stream);