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

Fixing EAGAIN processing for port message send.

Browse Source 
Sending large plain (exceeding port's max_size, not in shared memory) messages
causes message fragmentation.  First message fragment is sent successfully,
but the next fragment may fail with the EAGAIN error.  In this case, the
message has to be pushed back to queue head for additional processing.

Related to #167 issue on GitHub.
This commit is contained in:
Max Romanov
2019-03-05 15:38:50 +03:00
parent 0ab83370cb
commit 42b66ec654
1 changed files with 26 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
src/nxt_port_socket.c
View File

@@ -195,7 +195,24 @@ nxt_port_msg_create(nxt_task_t *task, nxt_port_send_msg_t *m)
static nxt_port_send_msg_t *
nxt_port_msg_push(nxt_task_t *task, nxt_port_t *port, nxt_port_send_msg_t *msg)
nxt_port_msg_insert_head(nxt_task_t *task, nxt_port_t *port,
nxt_port_send_msg_t *msg)
{
if (msg->work.data == NULL) {
msg = nxt_port_msg_create(task, msg);
}
if (msg != NULL) {
nxt_queue_insert_head(&port->messages, &msg->link);
}
return msg;
}
static nxt_port_send_msg_t *
nxt_port_msg_insert_tail(nxt_task_t *task, nxt_port_t *port,
nxt_port_send_msg_t *msg)
{
if (msg->work.data == NULL) {
msg = nxt_port_msg_create(task, msg);
@@ -260,7 +277,7 @@ nxt_port_socket_twrite(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
} else {
nxt_thread_mutex_lock(&port->write_mutex);
res = nxt_port_msg_push(task, port, &msg);
res = nxt_port_msg_insert_tail(task, port, &msg);
nxt_thread_mutex_unlock(&port->write_mutex);
@@ -409,7 +426,7 @@ nxt_port_write_handler(nxt_task_t *task, void *obj, void *data)
}
data = NULL;
if (nxt_port_msg_push(task, port, msg) != NULL) {
if (nxt_port_msg_insert_tail(task, port, msg) != NULL) {
use_delta++;
}
}
@@ -424,16 +441,17 @@ nxt_port_write_handler(nxt_task_t *task, void *obj, void *data)
data = NULL;
}
} else if (nxt_slow_path(n == NXT_ERROR)) {
} else {
if (msg->link.next == NULL) {
if (nxt_port_msg_push(task, port, msg) != NULL) {
if (nxt_port_msg_insert_head(task, port, msg) != NULL) {
use_delta++;
}
}
if (nxt_slow_path(n == NXT_ERROR)) {
goto fail;
}
/* n == NXT_AGAIN */
}
} while (port->socket.write_ready);