Extruded nxt_port_send_port() to send port to one process.
This commit is contained in:
@@ -81,13 +81,10 @@ void
|
||||
nxt_port_send_new_port(nxt_task_t *task, nxt_runtime_t *rt,
|
||||
nxt_port_t *new_port)
|
||||
{
|
||||
nxt_buf_t *b;
|
||||
nxt_port_t *port;
|
||||
nxt_process_t *process;
|
||||
nxt_port_msg_new_port_t *msg;
|
||||
nxt_process_t *process;
|
||||
|
||||
nxt_debug(task, "new port %d for process %PI engine %uD",
|
||||
new_port->socket.fd, new_port->pid, new_port->engine);
|
||||
new_port->pair[1], new_port->pid, new_port->engine);
|
||||
|
||||
nxt_runtime_process_each(rt, process)
|
||||
{
|
||||
@@ -95,36 +92,47 @@ nxt_port_send_new_port(nxt_task_t *task, nxt_runtime_t *rt,
|
||||
continue;
|
||||
}
|
||||
|
||||
port = nxt_process_port_first(process);
|
||||
|
||||
b = nxt_buf_mem_alloc(port->mem_pool, sizeof(nxt_port_data_t), 0);
|
||||
|
||||
if (nxt_slow_path(b == NULL)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
nxt_debug(task, "send new port %FD to process %PI",
|
||||
new_port->socket.fd, process->pid);
|
||||
|
||||
b->data = port->mem_pool;
|
||||
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;
|
||||
|
||||
msg->id = new_port->id;
|
||||
msg->pid = new_port->pid;
|
||||
msg->engine = new_port->engine;
|
||||
msg->max_size = port->max_size;
|
||||
msg->max_share = port->max_share;
|
||||
msg->type = new_port->type;
|
||||
|
||||
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_NEW_PORT,
|
||||
new_port->socket.fd, 0, 0, b);
|
||||
(void) nxt_port_send_port(task, nxt_process_port_first(process),
|
||||
new_port);
|
||||
}
|
||||
nxt_runtime_process_loop;
|
||||
}
|
||||
|
||||
|
||||
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);
|
||||
if (nxt_slow_path(b == NULL)) {
|
||||
return NXT_ERROR;
|
||||
}
|
||||
|
||||
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;
|
||||
|
||||
msg->id = new_port->id;
|
||||
msg->pid = new_port->pid;
|
||||
msg->engine = new_port->engine;
|
||||
msg->max_size = port->max_size;
|
||||
msg->max_share = port->max_share;
|
||||
msg->type = new_port->type;
|
||||
|
||||
return nxt_port_socket_write(task, port, NXT_PORT_MSG_NEW_PORT,
|
||||
new_port->pair[1], 0, 0, b);
|
||||
}
|
||||
|
||||
|
||||
static void
|
||||
nxt_port_new_port_buf_completion(nxt_task_t *task, void *obj, void *data)
|
||||
{
|
||||
|
||||
@@ -116,6 +116,8 @@ void nxt_port_write(nxt_task_t *task, nxt_runtime_t *rt, nxt_uint_t type,
|
||||
nxt_fd_t fd, uint32_t stream, nxt_buf_t *b);
|
||||
void nxt_port_send_new_port(nxt_task_t *task, nxt_runtime_t *rt,
|
||||
nxt_port_t *port);
|
||||
nxt_int_t nxt_port_send_port(nxt_task_t *task, nxt_port_t *port,
|
||||
nxt_port_t *new_port);
|
||||
void nxt_port_change_log_file(nxt_task_t *task, nxt_runtime_t *rt,
|
||||
nxt_uint_t slot, nxt_fd_t fd);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user