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

refactoring

Browse Source
This commit is contained in:
Devaev Maxim
2018-09-30 18:00:21 +03:00
parent f18a3ef992
commit b42a6c4124
3 changed files with 130 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
src/http.c
View File

@@ -187,45 +187,42 @@ static void _http_callback_ping(struct evhttp_request *request, void *v_server)
static void _http_callback_snapshot(struct evhttp_request *request, void *v_server) { static void _http_callback_snapshot(struct evhttp_request *request, void *v_server) {
struct http_server_t *server = (struct http_server_t *)v_server; struct http_server_t *server = (struct http_server_t *)v_server;
struct evbuffer *buf; struct evbuffer *buf;
char x_timestamp_buf[64]; char time_buf[64];
PROCESS_HEAD_REQUEST; PROCESS_HEAD_REQUEST;
# define EXPOSED(_next) server->run->exposed->_next
assert((buf = evbuffer_new())); assert((buf = evbuffer_new()));
assert(!evbuffer_add(buf, (const void *)server->run->exposed->picture.data, server->run->exposed->picture.size)); assert(!evbuffer_add(buf, (const void *)EXPOSED(picture.data), EXPOSED(picture.size)));
ADD_HEADER("Access-Control-Allow-Origin:", "*"); ADD_HEADER("Access-Control-Allow-Origin:", "*");
ADD_HEADER("Cache-Control", "no-store, no-cache, must-revalidate, pre-check=0, post-check=0, max-age=0"); ADD_HEADER("Cache-Control", "no-store, no-cache, must-revalidate, pre-check=0, post-check=0, max-age=0");
ADD_HEADER("Pragma", "no-cache"); ADD_HEADER("Pragma", "no-cache");
ADD_HEADER("Expires", "Mon, 3 Jan 2000 12:34:56 GMT"); ADD_HEADER("Expires", "Mon, 3 Jan 2000 12:34:56 GMT");
sprintf(x_timestamp_buf, "%.06Lf", now_real_ms()); # define ADD_TIME_HEADER(_key, _value) \
ADD_HEADER("X-Timestamp", x_timestamp_buf); { sprintf(time_buf, "%.06Lf", _value); ADD_HEADER(_key, time_buf); }
ADD_TIME_HEADER("X-Timestamp", now_real_ms());
if (server->add_x_timings) { if (server->add_x_timings) {
sprintf(x_timestamp_buf, "%.06Lf", server->run->exposed->picture.grab_time); ADD_TIME_HEADER("X-UStreamer-Grab-Time", EXPOSED(picture.grab_time));
ADD_HEADER("X-UStreamer-Grab-Time", x_timestamp_buf); ADD_TIME_HEADER("X-UStreamer-Encode-Begin-Time", EXPOSED(picture.encode_begin_time));
ADD_TIME_HEADER("X-UStreamer-Encode-End-Time", EXPOSED(picture.encode_end_time));
sprintf(x_timestamp_buf, "%.06Lf", server->run->exposed->picture.encode_begin_time); ADD_TIME_HEADER("X-UStreamer-Expose-Begin-Time", EXPOSED(expose_begin_time));
ADD_HEADER("X-UStreamer-Encode-Begin-Time", x_timestamp_buf); ADD_TIME_HEADER("X-UStreamer-Expose-End-Time", EXPOSED(expose_end_time));
ADD_TIME_HEADER("X-UStreamer-Send-Time", now_monotonic_ms());
sprintf(x_timestamp_buf, "%.06Lf", server->run->exposed->picture.encode_end_time);
ADD_HEADER("X-UStreamer-Encode-End-Time", x_timestamp_buf);
sprintf(x_timestamp_buf, "%.06Lf", server->run->exposed->expose_begin_time);
ADD_HEADER("X-UStreamer-Expose-Begin-Time", x_timestamp_buf);
sprintf(x_timestamp_buf, "%.06Lf", server->run->exposed->expose_end_time);
ADD_HEADER("X-UStreamer-Expose-End-Time", x_timestamp_buf);
sprintf(x_timestamp_buf, "%.06Lf", now_monotonic_ms());
ADD_HEADER("X-UStreamer-Send-Time", x_timestamp_buf);
} }
# undef ADD_TIME_HEADER
ADD_HEADER("Content-Type", "image/jpeg"); ADD_HEADER("Content-Type", "image/jpeg");
evhttp_send_reply(request, HTTP_OK, "OK", buf); evhttp_send_reply(request, HTTP_OK, "OK", buf);
evbuffer_free(buf); evbuffer_free(buf);
# undef EXPOSED
} }
#undef ADD_HEADER #undef ADD_HEADER
@@ -281,10 +278,10 @@ static void _http_callback_stream(struct evhttp_request *request, void *v_server
#undef PROCESS_HEAD_REQUEST #undef PROCESS_HEAD_REQUEST
#define BOUNDARY "boundarydonotcross"
#define RN "\r\n"
static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_client) { static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_client) {
# define BOUNDARY "boundarydonotcross"
# define RN "\r\n"
struct stream_client_t *client = (struct stream_client_t *)v_client; struct stream_client_t *client = (struct stream_client_t *)v_client;
struct evbuffer *buf; struct evbuffer *buf;
@@ -305,12 +302,14 @@ static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_c
client->need_initial = false; client->need_initial = false;
} }
# define EXPOSED(_next) client->server->run->exposed->_next
assert(evbuffer_add_printf(buf, assert(evbuffer_add_printf(buf,
"Content-Type: image/jpeg" RN "Content-Type: image/jpeg" RN
"Content-Length: %lu" RN "Content-Length: %lu" RN
"X-Timestamp: %.06Lf" RN "X-Timestamp: %.06Lf" RN
"%s", "%s",
client->server->run->exposed->picture.size * sizeof(*client->server->run->exposed->picture.data), EXPOSED(picture.size) * sizeof(*EXPOSED(picture.data)),
now_real_ms(), (client->server->add_x_timings ? "" : RN) now_real_ms(), (client->server->add_x_timings ? "" : RN)
)); ));
if (client->server->add_x_timings) { if (client->server->add_x_timings) {
@@ -322,18 +321,18 @@ static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_c
"X-UStreamer-Expose-End-Time: %.06Lf" RN "X-UStreamer-Expose-End-Time: %.06Lf" RN
"X-UStreamer-Send-Time: %.06Lf" RN "X-UStreamer-Send-Time: %.06Lf" RN
RN, RN,
client->server->run->exposed->picture.grab_time, EXPOSED(picture.grab_time),
client->server->run->exposed->picture.encode_begin_time, EXPOSED(picture.encode_begin_time),
client->server->run->exposed->picture.encode_end_time, EXPOSED(picture.encode_end_time),
client->server->run->exposed->expose_begin_time, EXPOSED(expose_begin_time),
client->server->run->exposed->expose_end_time, EXPOSED(expose_end_time),
now_monotonic_ms() now_monotonic_ms()
)); ));
} }
assert(!evbuffer_add(buf, assert(!evbuffer_add(buf,
(void *)client->server->run->exposed->picture.data, (void *)EXPOSED(picture.data),
client->server->run->exposed->picture.size * sizeof(*client->server->run->exposed->picture.data) EXPOSED(picture.size) * sizeof(*EXPOSED(picture.data))
)); ));
assert(evbuffer_add_printf(buf, RN "--" BOUNDARY RN)); assert(evbuffer_add_printf(buf, RN "--" BOUNDARY RN));
@@ -342,10 +341,11 @@ static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_c
bufferevent_setcb(buf_event, NULL, NULL, _http_callback_stream_error, (void *)client); bufferevent_setcb(buf_event, NULL, NULL, _http_callback_stream_error, (void *)client);
bufferevent_enable(buf_event, EV_READ); bufferevent_enable(buf_event, EV_READ);
}
#undef BOUNDARY # undef BOUNDARY
#undef RN # undef RN
# undef EXPOSED
}
static void _http_callback_stream_error(UNUSED struct bufferevent *buf_event, UNUSED short what, void *v_client) { static void _http_callback_stream_error(UNUSED struct bufferevent *buf_event, UNUSED short what, void *v_client) {
struct stream_client_t *client = (struct stream_client_t *)v_client; struct stream_client_t *client = (struct stream_client_t *)v_client;
@@ -431,87 +431,97 @@ static void _http_exposed_refresh(UNUSED int fd, UNUSED short what, void *v_serv
} }
static bool _expose_new_picture(struct http_server_t *server) { static bool _expose_new_picture(struct http_server_t *server) {
assert(server->run->stream->picture.size > 0); # define STREAM(_next) server->run->stream->_next
server->run->exposed->fps = server->run->stream->fps; # define EXPOSED(_next) server->run->exposed->_next
server->run->exposed->expose_begin_time = now_monotonic_ms();
assert(STREAM(picture.size) > 0);
EXPOSED(fps) = STREAM(fps);
EXPOSED(expose_begin_time) = now_monotonic_ms();
# define MEM_STREAM_TO_EXPOSED \ # define MEM_STREAM_TO_EXPOSED \
server->run->exposed->picture.data, server->run->stream->picture.data, \ EXPOSED(picture.data), STREAM(picture.data), \
server->run->stream->picture.size * sizeof(*server->run->exposed->picture.data) STREAM(picture.size) * sizeof(*STREAM(picture.data))
if (server->drop_same_frames) { if (server->drop_same_frames) {
if ( if (
server->run->exposed->online EXPOSED(online)
&& server->run->exposed->dropped < server->drop_same_frames && EXPOSED(dropped) < server->drop_same_frames
&& server->run->exposed->picture.size == server->run->stream->picture.size && EXPOSED(picture.size) == STREAM(picture.size)
&& !memcmp(MEM_STREAM_TO_EXPOSED) && !memcmp(MEM_STREAM_TO_EXPOSED)
) { ) {
LOG_PERF("HTTP: dropped same frame number %u", server->run->exposed->dropped); LOG_PERF("HTTP: dropped same frame number %u", EXPOSED(dropped));
server->run->exposed->dropped += 1; EXPOSED(dropped) += 1;
server->run->exposed->expose_end_time = now_monotonic_ms(); EXPOSED(expose_end_time) = now_monotonic_ms();
return false; // Not updated return false; // Not updated
} }
} }
if (server->run->exposed->picture.allocated < server->run->stream->picture.allocated) { if (EXPOSED(picture.allocated) < STREAM(picture.allocated)) {
A_REALLOC(server->run->exposed->picture.data, server->run->stream->picture.allocated); A_REALLOC(EXPOSED(picture.data), STREAM(picture.allocated));
server->run->exposed->picture.allocated = server->run->stream->picture.allocated; EXPOSED(picture.allocated) = STREAM(picture.allocated);
} }
memcpy(MEM_STREAM_TO_EXPOSED); memcpy(MEM_STREAM_TO_EXPOSED);
# undef MEM_STREAM_TO_EXPOSED # undef MEM_STREAM_TO_EXPOSED
server->run->exposed->picture.size = server->run->stream->picture.size; EXPOSED(picture.size) = STREAM(picture.size);
server->run->exposed->picture.grab_time = server->run->stream->picture.grab_time; EXPOSED(picture.grab_time) = STREAM(picture.grab_time);
server->run->exposed->picture.encode_begin_time = server->run->stream->picture.encode_begin_time; EXPOSED(picture.encode_begin_time) = STREAM(picture.encode_begin_time);
server->run->exposed->picture.encode_end_time = server->run->stream->picture.encode_end_time; EXPOSED(picture.encode_end_time) = STREAM(picture.encode_end_time);
server->run->exposed->width = server->run->stream->width; EXPOSED(width) = STREAM(width);
server->run->exposed->height = server->run->stream->height; EXPOSED(height) = STREAM(height);
server->run->exposed->online = true; EXPOSED(online) = true;
server->run->exposed->dropped = 0; EXPOSED(dropped) = 0;
server->run->exposed->expose_end_time = now_monotonic_ms(); EXPOSED(expose_end_time) = now_monotonic_ms();
# undef STREAM
# undef EXPOSED
return true; // Updated return true; // Updated
} }
static bool _expose_blank_picture(struct http_server_t *server) { static bool _expose_blank_picture(struct http_server_t *server) {
server->run->exposed->expose_begin_time = now_monotonic_ms(); # define EXPOSED(_next) server->run->exposed->_next
if (server->run->exposed->online || server->run->exposed->picture.size == 0) { EXPOSED(expose_begin_time) = now_monotonic_ms();
if (server->run->exposed->picture.allocated < BLANK_JPG_SIZE) {
A_REALLOC(server->run->exposed->picture.data, BLANK_JPG_SIZE); if (EXPOSED(online) || EXPOSED(picture.size) == 0) {
server->run->exposed->picture.allocated = BLANK_JPG_SIZE; if (EXPOSED(picture.allocated) < BLANK_JPG_SIZE) {
A_REALLOC(EXPOSED(picture.data), BLANK_JPG_SIZE);
EXPOSED(picture.allocated) = BLANK_JPG_SIZE;
} }
memcpy( memcpy(
server->run->exposed->picture.data, BLANK_JPG_DATA, EXPOSED(picture.data), BLANK_JPG_DATA,
BLANK_JPG_SIZE * sizeof(*server->run->exposed->picture.data) BLANK_JPG_SIZE * sizeof(*EXPOSED(picture.data))
); );
server->run->exposed->picture.size = BLANK_JPG_SIZE; EXPOSED(picture.size) = BLANK_JPG_SIZE;
server->run->exposed->picture.grab_time = 0; EXPOSED(picture.grab_time) = 0;
server->run->exposed->picture.encode_begin_time = 0; EXPOSED(picture.encode_begin_time) = 0;
server->run->exposed->picture.encode_end_time = 0; EXPOSED(picture.encode_end_time) = 0;
server->run->exposed->width = BLANK_JPG_WIDTH; EXPOSED(width) = BLANK_JPG_WIDTH;
server->run->exposed->height = BLANK_JPG_HEIGHT; EXPOSED(height) = BLANK_JPG_HEIGHT;
server->run->exposed->fps = 0; EXPOSED(fps) = 0;
server->run->exposed->online = false; EXPOSED(online) = false;
goto updated; goto updated;
} }
if (server->run->exposed->dropped < server->run->drop_same_frames_blank) { if (EXPOSED(dropped) < server->run->drop_same_frames_blank) {
LOG_PERF("HTTP: dropped same frame (BLANK) number %u", server->run->exposed->dropped); LOG_PERF("HTTP: dropped same frame (BLANK) number %u", EXPOSED(dropped));
server->run->exposed->dropped += 1; EXPOSED(dropped) += 1;
server->run->exposed->expose_end_time = now_monotonic_ms(); EXPOSED(expose_end_time) = now_monotonic_ms();
return false; // Not updated return false; // Not updated
} }
updated: updated:
server->run->exposed->dropped = 0; EXPOSED(dropped) = 0;
server->run->exposed->expose_end_time = now_monotonic_ms(); EXPOSED(expose_end_time) = now_monotonic_ms();
return true; // Updated return true; // Updated
# undef EXPOSED
} }

28
src/main.c
View File

@@ -135,11 +135,8 @@ static void _help(struct device_t *dev, struct encoder_t *encoder, struct http_s
} }
static int _parse_options(int argc, char *argv[], struct device_t *dev, struct encoder_t *encoder, struct http_server_t *server) { static int _parse_options(int argc, char *argv[], struct device_t *dev, struct encoder_t *encoder, struct http_server_t *server) {
# define OPT_ARG(_dest) \ # define OPT_SET(_dest, _value) \
{ _dest = optarg; break; } { _dest = _value; break; }
# define OPT_TRUE(_dest) \
{ _dest = true; break; }
# define OPT_UNSIGNED(_dest, _name, _min, _max) \ # define OPT_UNSIGNED(_dest, _name, _min, _max) \
{ errno = 0; int _tmp = strtol(optarg, NULL, 0); \ { errno = 0; int _tmp = strtol(optarg, NULL, 0); \
@@ -158,7 +155,7 @@ static int _parse_options(int argc, char *argv[], struct device_t *dev, struct e
log_level = LOG_LEVEL_INFO; log_level = LOG_LEVEL_INFO;
while ((ch = getopt_long(argc, argv, _short_opts, _long_opts, &index)) >= 0) { while ((ch = getopt_long(argc, argv, _short_opts, _long_opts, &index)) >= 0) {
switch (ch) { switch (ch) {
case 'd': OPT_ARG(dev->path); case 'd': OPT_SET(dev->path, optarg);
case 'x': OPT_UNSIGNED(dev->width, "--width", 320, 1920); case 'x': OPT_UNSIGNED(dev->width, "--width", 320, 1920);
case 'y': OPT_UNSIGNED(dev->height, "--height", 180, 1200); case 'y': OPT_UNSIGNED(dev->height, "--height", 180, 1200);
# pragma GCC diagnostic ignored "-Wsign-compare" # pragma GCC diagnostic ignored "-Wsign-compare"
@@ -168,38 +165,37 @@ static int _parse_options(int argc, char *argv[], struct device_t *dev, struct e
case 'a': OPT_PARSE(dev->standard, device_parse_standard, STANDARD_UNKNOWN, "TV standard"); case 'a': OPT_PARSE(dev->standard, device_parse_standard, STANDARD_UNKNOWN, "TV standard");
case 'e': OPT_UNSIGNED(dev->every_frame, "--every-frame", 1, 30); case 'e': OPT_UNSIGNED(dev->every_frame, "--every-frame", 1, 30);
case 'z': OPT_UNSIGNED(dev->min_frame_size, "--min-frame-size", 0, 8192); case 'z': OPT_UNSIGNED(dev->min_frame_size, "--min-frame-size", 0, 8192);
case 't': OPT_TRUE(dev->dv_timings); case 't': OPT_SET(dev->dv_timings, true);
case 'b': OPT_UNSIGNED(dev->n_buffers, "--buffers", 1, 32); case 'b': OPT_UNSIGNED(dev->n_buffers, "--buffers", 1, 32);
case 'w': OPT_UNSIGNED(dev->n_workers, "--workers", 1, 32); case 'w': OPT_UNSIGNED(dev->n_workers, "--workers", 1, 32);
case 'q': OPT_UNSIGNED(encoder->quality, "--quality", 1, 100); case 'q': OPT_UNSIGNED(encoder->quality, "--quality", 1, 100);
case 'c': OPT_PARSE(encoder->type, encoder_parse_type, ENCODER_TYPE_UNKNOWN, "encoder type"); case 'c': OPT_PARSE(encoder->type, encoder_parse_type, ENCODER_TYPE_UNKNOWN, "encoder type");
# ifdef OMX_ENCODER # ifdef OMX_ENCODER
case 500: encoder->omx_use_ijg = true; break; case 500: OPT_SET(encoder->omx_use_ijg, true);
# endif # endif
case 1000: OPT_UNSIGNED(dev->timeout, "--timeout", 1, 60); case 1000: OPT_UNSIGNED(dev->timeout, "--timeout", 1, 60);
case 1001: OPT_UNSIGNED(dev->error_timeout, "--error-timeout", 1, 60); case 1001: OPT_UNSIGNED(dev->error_timeout, "--error-timeout", 1, 60);
case 's': server->host = optarg; break; case 's': OPT_SET(server->host, optarg);
case 'p': OPT_UNSIGNED(server->port, "--port", 1, 65535); case 'p': OPT_UNSIGNED(server->port, "--port", 1, 65535);
case 'r': OPT_UNSIGNED(server->drop_same_frames, "--drop-same-frames", 0, 30); case 'r': OPT_UNSIGNED(server->drop_same_frames, "--drop-same-frames", 0, 30);
case 2000: server->add_x_timings = true; break; case 2000: OPT_SET(server->add_x_timings, true);
case 2001: OPT_UNSIGNED(server->fake_width, "--fake-width", 0, 1920); case 2001: OPT_UNSIGNED(server->fake_width, "--fake-width", 0, 1920);
case 2002: OPT_UNSIGNED(server->fake_height, "--fake-height", 0, 1200); case 2002: OPT_UNSIGNED(server->fake_height, "--fake-height", 0, 1200);
case 2003: OPT_UNSIGNED(server->timeout, "--server-timeout", 1, 60); case 2003: OPT_UNSIGNED(server->timeout, "--server-timeout", 1, 60);
case 5000: log_level = LOG_LEVEL_PERF; break; case 5000: OPT_SET(log_level, LOG_LEVEL_PERF);
case 5001: log_level = LOG_LEVEL_VERBOSE; break; case 5001: OPT_SET(log_level, LOG_LEVEL_VERBOSE);
case 5002: log_level = LOG_LEVEL_DEBUG; break; case 5002: OPT_SET(log_level, LOG_LEVEL_DEBUG);
case 5010: OPT_UNSIGNED(log_level, "--log-level", 0, 3); case 5010: OPT_UNSIGNED(log_level, "--log-level", 0, 3);
case 0: break; case 0: break;
case 'h': default: _help(dev, encoder, server); return -1; case 'h': default: _help(dev, encoder, server); return -1;
} }
} }
# undef OPT_PARSE # undef OPT_SET
# undef OPT_UNSIGNED # undef OPT_UNSIGNED
# undef OPT_TRUE # undef OPT_PARSE
# undef OPT_ARG
return 0; return 0;
} }

50
src/stream.c
View File

@@ -283,23 +283,27 @@ void stream_loop_break(struct stream_t *stream) {
} }
static void _stream_expose_picture(struct stream_t *stream, unsigned buf_index) { static void _stream_expose_picture(struct stream_t *stream, unsigned buf_index) {
# define PICTURE(_next) stream->dev->run->pictures[buf_index]._next
A_PTHREAD_M_LOCK(&stream->mutex); A_PTHREAD_M_LOCK(&stream->mutex);
stream->picture.size = stream->dev->run->pictures[buf_index].size; stream->picture.size = PICTURE(size);
stream->picture.allocated = stream->dev->run->pictures[buf_index].allocated; stream->picture.allocated = PICTURE(allocated);
memcpy( memcpy(
stream->picture.data, stream->dev->run->pictures[buf_index].data, stream->picture.data, PICTURE(data),
stream->picture.size * sizeof(*stream->picture.data) stream->picture.size * sizeof(*stream->picture.data)
); );
stream->picture.grab_time = stream->dev->run->pictures[buf_index].grab_time; stream->picture.grab_time = PICTURE(grab_time);
stream->picture.encode_begin_time = stream->dev->run->pictures[buf_index].encode_begin_time; stream->picture.encode_begin_time = PICTURE(encode_begin_time);
stream->picture.encode_end_time = stream->dev->run->pictures[buf_index].encode_end_time; stream->picture.encode_end_time = PICTURE(encode_end_time);
stream->width = stream->dev->run->width; stream->width = stream->dev->run->width;
stream->height = stream->dev->run->height; stream->height = stream->dev->run->height;
stream->updated = true; stream->updated = true;
A_PTHREAD_M_UNLOCK(&stream->mutex); A_PTHREAD_M_UNLOCK(&stream->mutex);
# undef PICTURE
} }
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) {
@@ -371,25 +375,29 @@ static void _stream_init_workers(struct device_t *dev, struct workers_pool_t *po
A_PTHREAD_M_INIT(&pool->workers[number].has_job_mutex); A_PTHREAD_M_INIT(&pool->workers[number].has_job_mutex);
A_PTHREAD_C_INIT(&pool->workers[number].has_job_cond); A_PTHREAD_C_INIT(&pool->workers[number].has_job_cond);
pool->workers[number].ctx.number = number; # define CTX(_next) pool->workers[number].ctx._next
pool->workers[number].ctx.dev = dev;
pool->workers[number].ctx.dev_stop = (sig_atomic_t *volatile)&dev->stop;
pool->workers[number].ctx.workers_stop = pool->workers_stop;
pool->workers[number].ctx.encoder = pool->encoder; CTX(number) = number;
CTX(dev) = dev;
CTX(dev_stop) = (sig_atomic_t *volatile)&dev->stop;
CTX(workers_stop) = pool->workers_stop;
pool->workers[number].ctx.last_comp_time_mutex = &pool->workers[number].last_comp_time_mutex; CTX(encoder) = pool->encoder;
pool->workers[number].ctx.last_comp_time = &pool->workers[number].last_comp_time;
pool->workers[number].ctx.has_job_mutex = &pool->workers[number].has_job_mutex; CTX(last_comp_time_mutex) = &pool->workers[number].last_comp_time_mutex;
pool->workers[number].ctx.has_job = &pool->workers[number].has_job; CTX(last_comp_time) = &pool->workers[number].last_comp_time;
pool->workers[number].ctx.job_failed = &pool->workers[number].job_failed;
pool->workers[number].ctx.job_start_time = &pool->workers[number].job_start_time;
pool->workers[number].ctx.has_job_cond = &pool->workers[number].has_job_cond;
pool->workers[number].ctx.free_workers_mutex = &pool->free_workers_mutex; CTX(has_job_mutex) = &pool->workers[number].has_job_mutex;
pool->workers[number].ctx.free_workers = &pool->free_workers; CTX(has_job) = &pool->workers[number].has_job;
pool->workers[number].ctx.free_workers_cond = &pool->free_workers_cond; CTX(job_failed) = &pool->workers[number].job_failed;
CTX(job_start_time) = &pool->workers[number].job_start_time;
CTX(has_job_cond) = &pool->workers[number].has_job_cond;
CTX(free_workers_mutex) = &pool->free_workers_mutex;
CTX(free_workers) = &pool->free_workers;
CTX(free_workers_cond) = &pool->free_workers_cond;
# undef CTX
A_PTHREAD_CREATE(&pool->workers[number].tid, _stream_worker_thread, (void *)&pool->workers[number].ctx); A_PTHREAD_CREATE(&pool->workers[number].tid, _stream_worker_thread, (void *)&pool->workers[number].ctx);
} }