technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Redirecting buffer completion handler to specific engine.

Browse Source 
There is a case in router where we use port in router connection thread.
Buffers are allocated within connection memory pool which can be used only in
this router thread. sendmsg() can be postponed into main router thread and
completion handler will compare current engine and post itself to correct
engine.
This commit is contained in:
Max Romanov
2017-07-07 16:01:34 +03:00
parent c9fbd832ab
commit f319220a6c
9 changed files with 239 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
src/nxt_buf.c
View File

@@ -8,13 +8,18 @@
static void nxt_buf_completion(nxt_task_t *task, void *obj, void *data);
static void nxt_buf_ts_completion(nxt_task_t *task, void *obj, void *data);
typedef struct {
nxt_work_t work;
nxt_event_engine_t *engine;
} nxt_buf_ts_t;
void
nxt_buf_mem_init(nxt_buf_t *b, void *start, size_t size)
{
b->size = NXT_BUF_MEM_SIZE;
b->mem.start = start;
b->mem.pos = start;
b->mem.free = start;
@@ -27,21 +32,18 @@ nxt_buf_mem_alloc(nxt_mp_t *mp, size_t size, nxt_uint_t flags)
{
nxt_buf_t *b;
b = nxt_mp_zalloc(mp, NXT_BUF_MEM_SIZE);
b = nxt_mp_alloc(mp, NXT_BUF_MEM_SIZE + size);
if (nxt_slow_path(b == NULL)) {
return NULL;
}
nxt_memzero(b, NXT_BUF_MEM_SIZE);
b->data = mp;
b->completion_handler = nxt_buf_completion;
b->size = NXT_BUF_MEM_SIZE;
if (size != 0) {
b->mem.start = nxt_mp_alloc(mp, size);
if (nxt_slow_path(b->mem.start == NULL)) {
return NULL;
}
b->mem.start = nxt_pointer_to(b, NXT_BUF_MEM_SIZE);
b->mem.pos = b->mem.start;
b->mem.free = b->mem.start;
b->mem.end = b->mem.start + size;
@@ -52,26 +54,60 @@ nxt_buf_mem_alloc(nxt_mp_t *mp, size_t size, nxt_uint_t flags)
nxt_buf_t *
nxt_buf_file_alloc(nxt_mp_t *mp, size_t size, nxt_uint_t flags)
nxt_buf_mem_ts_alloc(nxt_task_t *task, nxt_mp_t *mp, size_t size)
{
nxt_buf_t *b;
nxt_buf_t *b;
nxt_buf_ts_t *ts;
b = nxt_mp_zalloc(mp, NXT_BUF_FILE_SIZE);
b = nxt_mp_retain(mp, NXT_BUF_MEM_SIZE + sizeof(nxt_buf_ts_t) + size);
if (nxt_slow_path(b == NULL)) {
return NULL;
}
nxt_memzero(b, NXT_BUF_MEM_SIZE + sizeof(nxt_buf_ts_t));
b->data = mp;
b->completion_handler = nxt_buf_ts_completion;
b->is_ts = 1;
if (size != 0) {
b->mem.start = nxt_pointer_to(b, NXT_BUF_MEM_SIZE +
sizeof(nxt_buf_ts_t));
b->mem.pos = b->mem.start;
b->mem.free = b->mem.start;
b->mem.end = b->mem.start + size;
}
ts = nxt_pointer_to(b, NXT_BUF_MEM_SIZE);
ts->engine = task->thread->engine;
ts->work.handler = nxt_buf_ts_completion;
ts->work.task = task;
ts->work.obj = b;
ts->work.data = b->parent;
return b;
}
nxt_buf_t *
nxt_buf_file_alloc(nxt_mp_t *mp, size_t size, nxt_uint_t flags)
{
nxt_buf_t *b;
b = nxt_mp_alloc(mp, NXT_BUF_FILE_SIZE + size);
if (nxt_slow_path(b == NULL)) {
return NULL;
}
nxt_memzero(b, NXT_BUF_FILE_SIZE);
b->data = mp;
b->completion_handler = nxt_buf_completion;
b->size = NXT_BUF_FILE_SIZE;
nxt_buf_set_file(b);
if (size != 0) {
b->mem.start = nxt_mp_alloc(mp, size);
if (nxt_slow_path(b->mem.start == NULL)) {
return NULL;
}
b->mem.start = nxt_pointer_to(b, NXT_BUF_FILE_SIZE);
b->mem.pos = b->mem.start;
b->mem.free = b->mem.start;
b->mem.end = b->mem.start + size;
@@ -91,7 +127,6 @@ nxt_buf_mmap_alloc(nxt_mp_t *mp, size_t size)
if (nxt_fast_path(b != NULL)) {
b->data = mp;
b->completion_handler = nxt_buf_completion;
b->size = NXT_BUF_MMAP_SIZE;
nxt_buf_set_file(b);
nxt_buf_set_mmap(b);
@@ -112,7 +147,6 @@ nxt_buf_sync_alloc(nxt_mp_t *mp, nxt_uint_t flags)
if (nxt_fast_path(b != NULL)) {
b->data = mp;
b->completion_handler = nxt_buf_completion;
b->size = NXT_BUF_SYNC_SIZE;
nxt_buf_set_sync(b);
b->is_nobuf = ((flags & NXT_BUF_SYNC_NOBUF) != 0);
@@ -166,8 +200,91 @@ nxt_buf_completion(nxt_task_t *task, void *obj, void *data)
nxt_debug(task, "buf completion: %p %p", b, b->mem.start);
#if (NXT_DEBUG)
if (nxt_slow_path(data != b->parent)) {
nxt_log_alert(task->log, "completion data (%p) != b->parent (%p)",
data, b->parent);
nxt_abort();
}
#endif
mp = b->data;
nxt_buf_free(mp, b);
nxt_mp_free(mp, b);
if (parent != NULL) {
nxt_debug(task, "parent retain:%uD", parent->retain);
parent->retain--;
if (parent->retain == 0) {
parent->mem.pos = parent->mem.free;
parent->completion_handler(task, parent, parent->parent);
}
}
}
nxt_int_t
nxt_buf_ts_handle(nxt_task_t *task, void *obj, void *data)
{
nxt_buf_t *b;
nxt_buf_ts_t *ts;
b = obj;
#if (NXT_DEBUG)
if (nxt_slow_path(b->is_ts == 0)) {
nxt_log_alert(task->log, "not a thread safe buf (%p) completed", b);
nxt_abort();
}
#endif
ts = nxt_pointer_to(b, NXT_BUF_MEM_SIZE);
if (ts->engine != task->thread->engine) {
nxt_debug(task, "buf ts: %p current engine is %p, expected %p",
b, task->thread->engine, ts->engine);
ts->work.handler = b->completion_handler;
ts->work.obj = obj;
ts->work.data = data;
nxt_event_engine_post(ts->engine, &ts->work);
return 1;
}
return 0;
}
static void
nxt_buf_ts_completion(nxt_task_t *task, void *obj, void *data)
{
nxt_mp_t *mp;
nxt_buf_t *b, *parent;
b = obj;
parent = data;
if (nxt_buf_ts_handle(task, obj, data)) {
return;
}
nxt_debug(task, "buf ts completion: %p %p", b, b->mem.start);
#if (NXT_DEBUG)
if (nxt_slow_path(data != b->parent)) {
nxt_log_alert(task->log, "completion data (%p) != b->parent (%p)",
data, b->parent);
nxt_abort();
}
#endif
mp = b->data;
nxt_mp_release(mp, b);
if (parent != NULL) {
nxt_debug(task, "parent retain:%uD", parent->retain);

9
src/nxt_buf.h
View File

@@ -76,11 +76,6 @@ struct nxt_buf_s {
nxt_buf_t *next;
uint32_t retain;
/*
* Used by nxt_mem_cache_free() to return buffer
* in appropriate memory pool cache.
*/
uint8_t size;
uint8_t is_file; /* 1 bit */
@@ -92,6 +87,7 @@ struct nxt_buf_s {
uint16_t is_flush:1;
uint16_t is_last:1;
uint16_t is_port_mmap_sent:1;
uint16_t is_ts:1;
nxt_buf_mem_t mem;
@@ -245,11 +241,14 @@ nxt_buf_used_size(b) \
NXT_EXPORT void nxt_buf_mem_init(nxt_buf_t *b, void *start, size_t size);
NXT_EXPORT nxt_buf_t *nxt_buf_mem_alloc(nxt_mp_t *mp, size_t size,
nxt_uint_t flags);
NXT_EXPORT nxt_buf_t *nxt_buf_mem_ts_alloc(nxt_task_t *task, nxt_mp_t *mp,
size_t size);
NXT_EXPORT nxt_buf_t *nxt_buf_file_alloc(nxt_mp_t *mp, size_t size,
nxt_uint_t flags);
NXT_EXPORT nxt_buf_t *nxt_buf_mmap_alloc(nxt_mp_t *mp, size_t size);
NXT_EXPORT nxt_buf_t *nxt_buf_sync_alloc(nxt_mp_t *mp, nxt_uint_t flags);
NXT_EXPORT nxt_int_t nxt_buf_ts_handle(nxt_task_t *task, void *obj, void *data);
#define \
nxt_buf_free(mp, b) \

7
src/nxt_buf_pool.c
View File

@@ -147,10 +147,6 @@ nxt_buf_pool_free(nxt_buf_pool_t *bp, nxt_buf_t *b)
bp->current = NULL;
}
if (!bp->mmap) {
nxt_mp_free(bp->mem_pool, b->mem.start);
}
nxt_buf_free(bp->mem_pool, b);
return;
@@ -176,15 +172,12 @@ nxt_buf_pool_free(nxt_buf_pool_t *bp, nxt_buf_t *b)
void
nxt_buf_pool_destroy(nxt_buf_pool_t *bp)
{
u_char *p;
nxt_buf_t *b;
bp->destroy = 1;
for (b = bp->free; b != NULL; b = b->next) {
p = b->mem.start;
nxt_buf_free(bp->mem_pool, b);
nxt_mp_free(bp->mem_pool, p);
}
bp->free = b; /* NULL */

3
src/nxt_log.h
View File

@@ -122,7 +122,8 @@ nxt_log_debug(_log, ...) \
} while (0)
#define nxt_debug_alloc(...)
#define nxt_debug_alloc(_fmt, ARGS...) \
nxt_thread_log_debug("%08p: " _fmt, mp, ##ARGS)
#else

22
src/nxt_port.c
View File

@@ -10,8 +10,6 @@
static void nxt_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
static void nxt_port_new_port_buf_completion(nxt_task_t *task, void *obj,
void *data);
void
@@ -102,13 +100,10 @@ nxt_port_send_new_port(nxt_task_t *task, nxt_runtime_t *rt,
nxt_int_t
nxt_port_send_port(nxt_task_t *task, nxt_port_t *port, nxt_port_t *new_port)
{
nxt_mp_t *mp;
nxt_buf_t *b;
nxt_port_msg_new_port_t *msg;
mp = port->mem_pool;
b = nxt_buf_mem_alloc(mp, sizeof(nxt_port_data_t), 0);
b = nxt_buf_mem_ts_alloc(task, port->mem_pool, sizeof(nxt_port_data_t));
if (nxt_slow_path(b == NULL)) {
return NXT_ERROR;
}
@@ -116,8 +111,6 @@ nxt_port_send_port(nxt_task_t *task, nxt_port_t *port, nxt_port_t *new_port)
nxt_debug(task, "send port %FD to process %PI",
new_port->pair[1], port->pid);
b->data = mp;
b->completion_handler = nxt_port_new_port_buf_completion;
b->mem.free += sizeof(nxt_port_msg_new_port_t);
msg = (nxt_port_msg_new_port_t *) b->mem.pos;
@@ -133,19 +126,6 @@ nxt_port_send_port(nxt_task_t *task, nxt_port_t *port, nxt_port_t *new_port)
}
static void
nxt_port_new_port_buf_completion(nxt_task_t *task, void *obj, void *data)
{
nxt_mp_t *mp;
nxt_buf_t *b;
b = obj;
mp = b->data;
nxt_buf_free(mp, b);
}
void
nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{

5
src/nxt_port.h
View File

@@ -36,10 +36,13 @@ typedef struct {
typedef struct {
nxt_queue_link_t link;
nxt_buf_t *buf;
nxt_mp_t *mem_pool;
size_t share;
nxt_fd_t fd;
nxt_port_msg_t port_msg;
nxt_work_t work;
nxt_event_engine_t *engine;
nxt_mp_t *mem_pool;
} nxt_port_send_msg_t;

29
src/nxt_port_memory.c
View File

@@ -35,6 +35,10 @@ nxt_port_mmap_buf_completion(nxt_task_t *task, void *obj, void *data)
nxt_chunk_id_t c;
nxt_port_mmap_header_t *hdr;
if (nxt_buf_ts_handle(task, obj, data)) {
return;
}
b = obj;
nxt_debug(task, "mmap buf completion: %p %p", b, b->mem.start);
@@ -71,7 +75,7 @@ nxt_port_mmap_buf_completion(nxt_task_t *task, void *obj, void *data)
c++;
}
nxt_buf_free(mp, b);
nxt_mp_release(mp, b);
}
@@ -150,6 +154,10 @@ nxt_port_mmap_send_fd_buf_completion(nxt_task_t *task, void *obj, void *data)
nxt_buf_t *b;
nxt_mp_t *mp;
if (nxt_buf_ts_handle(task, obj, data)) {
return;
}
b = obj;
mp = b->data;
fd = (nxt_fd_t) (intptr_t) data;
@@ -166,7 +174,7 @@ nxt_port_mmap_send_fd_buf_completion(nxt_task_t *task, void *obj, void *data)
nxt_fd_close(fd);
nxt_buf_free(mp, b);
nxt_mp_release(mp, b);
}
@@ -252,13 +260,12 @@ nxt_port_new_port_mmap(nxt_task_t *task, nxt_process_t *process,
port_mmap->hdr = mem;
b = nxt_buf_mem_alloc(port->mem_pool, 0, 0);
b = nxt_buf_mem_ts_alloc(task, port->mem_pool, 0);
if (nxt_slow_path(b == NULL)) {
goto remove_fail;
}
b->completion_handler = nxt_port_mmap_send_fd_buf_completion;
b->data = port->mem_pool;
b->parent = (void *) (intptr_t) fd;
/* Init segment header. */
@@ -396,24 +403,22 @@ nxt_port_mmap_get_buf(nxt_task_t *task, nxt_port_t *port, size_t size)
return NULL;
}
b = nxt_mp_zalloc(port->mem_pool, NXT_BUF_PORT_MMAP_SIZE);
b = nxt_buf_mem_ts_alloc(task, port->mem_pool, 0);
if (nxt_slow_path(b == NULL)) {
return NULL;
}
b->data = port->mem_pool;
b->completion_handler = nxt_port_mmap_buf_completion;
b->size = NXT_BUF_PORT_MMAP_SIZE;
nxt_buf_set_port_mmap(b);
hdr = nxt_port_mmap_get(task, port, &c, size);
if (nxt_slow_path(hdr == NULL)) {
nxt_buf_free(port->mem_pool, b);
nxt_mp_release(port->mem_pool, b);
return NULL;
}
b->parent = hdr;
b->mem.start = nxt_port_mmap_chunk_start(hdr, c);
b->mem.pos = b->mem.start;
b->mem.free = b->mem.start;
@@ -517,14 +522,12 @@ nxt_port_mmap_get_incoming_buf(nxt_task_t *task, nxt_port_t *port,
return NULL;
}
b = nxt_mp_zalloc(port->mem_pool, NXT_BUF_PORT_MMAP_SIZE);
b = nxt_buf_mem_ts_alloc(task, port->mem_pool, 0);
if (nxt_slow_path(b == NULL)) {
return NULL;
}
b->data = port->mem_pool;
b->completion_handler = nxt_port_mmap_buf_completion;
b->size = NXT_BUF_PORT_MMAP_SIZE;
nxt_buf_set_port_mmap(b);
@@ -559,7 +562,7 @@ nxt_port_mmap_write(nxt_task_t *task, nxt_port_t *port,
bsize = sb->niov * sizeof(nxt_port_mmap_msg_t);
b = nxt_buf_mem_alloc(port->mem_pool, bsize, 0);
b = nxt_buf_mem_ts_alloc(task, port->mem_pool, bsize);
if (nxt_slow_path(b == NULL)) {
return;
}

46
src/nxt_port_socket.c
View File

@@ -135,6 +135,37 @@ nxt_port_write_close(nxt_port_t *port)
}
static void
nxt_port_release_send_msg(nxt_task_t *task, void *obj, void *data)
{
nxt_event_engine_t *engine;
nxt_port_send_msg_t *msg;
msg = obj;
engine = data;
#if (NXT_DEBUG)
if (nxt_slow_path(data != msg->engine)) {
nxt_log_alert(task->log, "release msg data (%p) != msg->engine (%p)",
data, msg->engine);
nxt_abort();
}
#endif
if (engine != task->thread->engine) {
nxt_debug(task, "current thread is %PT, expected %PT",
task->thread->tid, engine->task.thread->tid);
nxt_event_engine_post(engine, &msg->work);
return;
}
nxt_mp_release(msg->mem_pool, obj);
}
nxt_int_t
nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
nxt_fd_t fd, uint32_t stream, nxt_port_id_t reply_port, nxt_buf_t *b)
@@ -160,14 +191,25 @@ nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
}
}
msg = nxt_mp_zalloc(port->mem_pool, sizeof(nxt_port_send_msg_t));
msg = nxt_mp_retain(port->mem_pool, sizeof(nxt_port_send_msg_t));
if (nxt_slow_path(msg == NULL)) {
return NXT_ERROR;
}
msg->link.next = NULL;
msg->link.prev = NULL;
msg->buf = b;
msg->fd = fd;
msg->share = 0;
msg->work.next = NULL;
msg->work.handler = nxt_port_release_send_msg;
msg->work.task = task;
msg->work.obj = msg;
msg->work.data = task->thread->engine;
msg->engine = task->thread->engine;
msg->mem_pool = port->mem_pool;
msg->port_msg.stream = stream;
@@ -292,7 +334,7 @@ nxt_port_write_handler(nxt_task_t *task, void *obj, void *data)
} else {
nxt_queue_remove(link);
nxt_mp_free(msg->mem_pool, msg);
nxt_port_release_send_msg(task, msg, msg->engine);
}
} else if (nxt_slow_path(n == NXT_ERROR)) {

41
src/nxt_router.c
View File

@@ -788,7 +788,6 @@ static nxt_int_t
nxt_router_thread_create(nxt_task_t *task, nxt_runtime_t *rt,
nxt_event_engine_t *engine)
{
nxt_mp_t *mp;
nxt_int_t ret;
nxt_port_t *port;
nxt_process_t *process;
@@ -824,12 +823,6 @@ nxt_router_thread_create(nxt_task_t *task, nxt_runtime_t *rt,
return ret;
}
mp = nxt_mp_create(1024, 128, 256, 32);
if (nxt_slow_path(mp == NULL)) {
return NXT_ERROR;
}
port->mem_pool = mp;
port->engine = 0;
port->type = NXT_PROCESS_ROUTER;
@@ -1441,6 +1434,8 @@ static void
nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
nxt_app_parse_ctx_t *ap)
{
nxt_mp_t *port_mp;
nxt_int_t res;
nxt_port_t *port, *c_port;
nxt_req_id_t req_id;
nxt_app_wmsg_t wmsg;
@@ -1466,6 +1461,7 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
} while (nxt_event_engine_request_find(engine, req_id) != NULL);
rc = nxt_conn_request_add(c, req_id);
if (nxt_slow_path(rc == NULL)) {
// 500 Failed to allocate req->conn link
nxt_log_alert(task->log, "failed to allocate req->conn link");
@@ -1476,11 +1472,20 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
nxt_debug(task, "req_id %uxD linked to conn %p at engine %p",
req_id, c, engine);
port_mp = port->mem_pool;
port->mem_pool = c->mem_pool;
c_port = nxt_process_connected_port_find(port->process,
engine->port->pid,
engine->port->id);
if (nxt_slow_path(c_port != engine->port)) {
(void) nxt_port_send_port(task, port, engine->port);
res = nxt_port_send_port(task, port, engine->port);
if (nxt_slow_path(res != NXT_OK)) {
// 500 Failed to send reply port
nxt_log_alert(task->log, "failed to send reply port to application");
}
nxt_process_connected_port_add(port->process, engine->port);
}
@@ -1489,14 +1494,26 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
wmsg.buf = &wmsg.write;
wmsg.stream = req_id;
(void)nxt_app->prepare_msg(task, &ap->r, &wmsg);
res = nxt_app->prepare_msg(task, &ap->r, &wmsg);
if (nxt_slow_path(res != NXT_OK)) {
// 500 Failed to prepare message
nxt_log_alert(task->log, "failed to prepare message for application");
}
nxt_debug(task, "about to send %d bytes buffer to worker port %d",
nxt_buf_used_size(wmsg.write),
wmsg.port->socket.fd);
(void) nxt_port_socket_write(task, wmsg.port, NXT_PORT_MSG_DATA,
res = nxt_port_socket_write(task, wmsg.port, NXT_PORT_MSG_DATA,
-1, req_id, engine->port->id, wmsg.write);
if (nxt_slow_path(res != NXT_OK)) {
// 500 Failed to send message
nxt_log_alert(task->log, "failed to send message to application");
}
port->mem_pool = port_mp;
}
@@ -1597,7 +1614,9 @@ nxt_router_conn_free(nxt_task_t *task, void *obj, void *data)
} nxt_queue_loop;
nxt_mp_destroy(c->mem_pool);
nxt_queue_remove(&c->link);
nxt_mp_release(c->mem_pool, c);
}