xref: /unit/src/nxt_port.c (revision 2050:d1298cc3f385)
111Sigor@sysoev.ru 
211Sigor@sysoev.ru /*
311Sigor@sysoev.ru  * Copyright (C) Igor Sysoev
411Sigor@sysoev.ru  * Copyright (C) NGINX, Inc.
511Sigor@sysoev.ru  */
611Sigor@sysoev.ru 
711Sigor@sysoev.ru #include <nxt_main.h>
820Sigor@sysoev.ru #include <nxt_runtime.h>
911Sigor@sysoev.ru #include <nxt_port.h>
10163Smax.romanov@nginx.com #include <nxt_router.h>
111789Smax.romanov@nginx.com #include <nxt_app_queue.h>
121555Smax.romanov@nginx.com #include <nxt_port_queue.h>
1311Sigor@sysoev.ru 
1411Sigor@sysoev.ru 
151998St.nateldemoura@f5.com static void nxt_port_remove_pid(nxt_task_t *task, nxt_port_recv_msg_t *msg,
161998St.nateldemoura@f5.com     nxt_pid_t pid);
1714Sigor@sysoev.ru static void nxt_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
1811Sigor@sysoev.ru 
19163Smax.romanov@nginx.com static nxt_atomic_uint_t nxt_port_last_id = 1;
20163Smax.romanov@nginx.com 
21163Smax.romanov@nginx.com 
22197Smax.romanov@nginx.com static void
nxt_port_mp_cleanup(nxt_task_t * task,void * obj,void * data)23197Smax.romanov@nginx.com nxt_port_mp_cleanup(nxt_task_t *task, void *obj, void *data)
24197Smax.romanov@nginx.com {
25197Smax.romanov@nginx.com     nxt_mp_t    *mp;
26197Smax.romanov@nginx.com     nxt_port_t  *port;
27197Smax.romanov@nginx.com 
28197Smax.romanov@nginx.com     port = obj;
29197Smax.romanov@nginx.com     mp = data;
30197Smax.romanov@nginx.com 
31197Smax.romanov@nginx.com     nxt_assert(port->pair[0] == -1);
32197Smax.romanov@nginx.com     nxt_assert(port->pair[1] == -1);
33197Smax.romanov@nginx.com 
34343Smax.romanov@nginx.com     nxt_assert(port->use_count == 0);
35197Smax.romanov@nginx.com     nxt_assert(port->app_link.next == NULL);
36507Smax.romanov@nginx.com     nxt_assert(port->idle_link.next == NULL);
37197Smax.romanov@nginx.com 
38197Smax.romanov@nginx.com     nxt_assert(nxt_queue_is_empty(&port->messages));
39197Smax.romanov@nginx.com     nxt_assert(nxt_lvlhsh_is_empty(&port->rpc_streams));
40197Smax.romanov@nginx.com     nxt_assert(nxt_lvlhsh_is_empty(&port->rpc_peers));
41197Smax.romanov@nginx.com 
42343Smax.romanov@nginx.com     nxt_thread_mutex_destroy(&port->write_mutex);
43343Smax.romanov@nginx.com 
44197Smax.romanov@nginx.com     nxt_mp_free(mp, port);
45197Smax.romanov@nginx.com }
46197Smax.romanov@nginx.com 
47197Smax.romanov@nginx.com 
48163Smax.romanov@nginx.com nxt_port_t *
nxt_port_new(nxt_task_t * task,nxt_port_id_t id,nxt_pid_t pid,nxt_process_type_t type)49197Smax.romanov@nginx.com nxt_port_new(nxt_task_t *task, nxt_port_id_t id, nxt_pid_t pid,
50197Smax.romanov@nginx.com     nxt_process_type_t type)
51163Smax.romanov@nginx.com {
52163Smax.romanov@nginx.com     nxt_mp_t    *mp;
53163Smax.romanov@nginx.com     nxt_port_t  *port;
54163Smax.romanov@nginx.com 
55163Smax.romanov@nginx.com     mp = nxt_mp_create(1024, 128, 256, 32);
56163Smax.romanov@nginx.com 
57163Smax.romanov@nginx.com     if (nxt_slow_path(mp == NULL)) {
58163Smax.romanov@nginx.com         return NULL;
59163Smax.romanov@nginx.com     }
60163Smax.romanov@nginx.com 
61163Smax.romanov@nginx.com     port = nxt_mp_zalloc(mp, sizeof(nxt_port_t));
62163Smax.romanov@nginx.com 
63163Smax.romanov@nginx.com     if (nxt_fast_path(port != NULL)) {
64163Smax.romanov@nginx.com         port->id = id;
65163Smax.romanov@nginx.com         port->pid = pid;
66163Smax.romanov@nginx.com         port->type = type;
67163Smax.romanov@nginx.com         port->mem_pool = mp;
68343Smax.romanov@nginx.com         port->use_count = 1;
69163Smax.romanov@nginx.com 
70197Smax.romanov@nginx.com         nxt_mp_cleanup(mp, nxt_port_mp_cleanup, task, port, mp);
71197Smax.romanov@nginx.com 
72163Smax.romanov@nginx.com         nxt_queue_init(&port->messages);
73343Smax.romanov@nginx.com         nxt_thread_mutex_create(&port->write_mutex);
74163Smax.romanov@nginx.com 
751555Smax.romanov@nginx.com         port->queue_fd = -1;
761555Smax.romanov@nginx.com 
77163Smax.romanov@nginx.com     } else {
78163Smax.romanov@nginx.com         nxt_mp_destroy(mp);
79163Smax.romanov@nginx.com     }
80163Smax.romanov@nginx.com 
81167Smax.romanov@nginx.com     nxt_thread_log_debug("port %p %d:%d new, type %d", port, pid, id, type);
82167Smax.romanov@nginx.com 
83163Smax.romanov@nginx.com     return port;
84163Smax.romanov@nginx.com }
85163Smax.romanov@nginx.com 
86163Smax.romanov@nginx.com 
87343Smax.romanov@nginx.com void
nxt_port_close(nxt_task_t * task,nxt_port_t * port)88343Smax.romanov@nginx.com nxt_port_close(nxt_task_t *task, nxt_port_t *port)
89163Smax.romanov@nginx.com {
901789Smax.romanov@nginx.com     size_t  size;
911789Smax.romanov@nginx.com 
92343Smax.romanov@nginx.com     nxt_debug(task, "port %p %d:%d close, type %d", port, port->pid,
93343Smax.romanov@nginx.com               port->id, port->type);
94167Smax.romanov@nginx.com 
95163Smax.romanov@nginx.com     if (port->pair[0] != -1) {
96583Smax.romanov@nginx.com         nxt_port_rpc_close(task, port);
97583Smax.romanov@nginx.com 
98163Smax.romanov@nginx.com         nxt_fd_close(port->pair[0]);
99163Smax.romanov@nginx.com         port->pair[0] = -1;
100163Smax.romanov@nginx.com     }
101163Smax.romanov@nginx.com 
102163Smax.romanov@nginx.com     if (port->pair[1] != -1) {
103163Smax.romanov@nginx.com         nxt_fd_close(port->pair[1]);
104163Smax.romanov@nginx.com         port->pair[1] = -1;
105343Smax.romanov@nginx.com 
106343Smax.romanov@nginx.com         if (port->app != NULL) {
107343Smax.romanov@nginx.com             nxt_router_app_port_close(task, port);
108343Smax.romanov@nginx.com         }
109163Smax.romanov@nginx.com     }
1101555Smax.romanov@nginx.com 
1111555Smax.romanov@nginx.com     if (port->queue_fd != -1) {
1121555Smax.romanov@nginx.com         nxt_fd_close(port->queue_fd);
1131555Smax.romanov@nginx.com         port->queue_fd = -1;
1141555Smax.romanov@nginx.com     }
1151555Smax.romanov@nginx.com 
1161555Smax.romanov@nginx.com     if (port->queue != NULL) {
1171789Smax.romanov@nginx.com         size = (port->id == (nxt_port_id_t) -1) ? sizeof(nxt_app_queue_t)
1181789Smax.romanov@nginx.com                                                 : sizeof(nxt_port_queue_t);
1191789Smax.romanov@nginx.com         nxt_mem_munmap(port->queue, size);
1201789Smax.romanov@nginx.com 
1211555Smax.romanov@nginx.com         port->queue = NULL;
1221555Smax.romanov@nginx.com     }
123343Smax.romanov@nginx.com }
124343Smax.romanov@nginx.com 
125163Smax.romanov@nginx.com 
126343Smax.romanov@nginx.com static void
nxt_port_release(nxt_task_t * task,nxt_port_t * port)127343Smax.romanov@nginx.com nxt_port_release(nxt_task_t *task, nxt_port_t *port)
128343Smax.romanov@nginx.com {
129343Smax.romanov@nginx.com     nxt_debug(task, "port %p %d:%d release, type %d", port, port->pid,
130343Smax.romanov@nginx.com               port->id, port->type);
131343Smax.romanov@nginx.com 
132753Smax.romanov@nginx.com     port->app = NULL;
133163Smax.romanov@nginx.com 
134163Smax.romanov@nginx.com     if (port->link.next != NULL) {
135349Smax.romanov@nginx.com         nxt_assert(port->process != NULL);
136349Smax.romanov@nginx.com 
137163Smax.romanov@nginx.com         nxt_process_port_remove(port);
138349Smax.romanov@nginx.com 
139349Smax.romanov@nginx.com         nxt_process_use(task, port->process, -1);
140163Smax.romanov@nginx.com     }
141163Smax.romanov@nginx.com 
142430Sigor@sysoev.ru     nxt_mp_release(port->mem_pool);
143163Smax.romanov@nginx.com }
144163Smax.romanov@nginx.com 
145141Smax.romanov@nginx.com 
146141Smax.romanov@nginx.com nxt_port_id_t
nxt_port_get_next_id()147141Smax.romanov@nginx.com nxt_port_get_next_id()
148141Smax.romanov@nginx.com {
149141Smax.romanov@nginx.com     return nxt_atomic_fetch_add(&nxt_port_last_id, 1);
150141Smax.romanov@nginx.com }
151141Smax.romanov@nginx.com 
15211Sigor@sysoev.ru 
15311Sigor@sysoev.ru void
nxt_port_reset_next_id()154141Smax.romanov@nginx.com nxt_port_reset_next_id()
155141Smax.romanov@nginx.com {
156141Smax.romanov@nginx.com     nxt_port_last_id = 1;
157141Smax.romanov@nginx.com }
158141Smax.romanov@nginx.com 
159141Smax.romanov@nginx.com 
160141Smax.romanov@nginx.com void
nxt_port_enable(nxt_task_t * task,nxt_port_t * port,const nxt_port_handlers_t * handlers)161141Smax.romanov@nginx.com nxt_port_enable(nxt_task_t *task, nxt_port_t *port,
1621302St.nateldemoura@f5.com     const nxt_port_handlers_t *handlers)
16311Sigor@sysoev.ru {
16414Sigor@sysoev.ru     port->pid = nxt_pid;
16514Sigor@sysoev.ru     port->handler = nxt_port_handler;
166320Smax.romanov@nginx.com     port->data = (nxt_port_handler_t *) (handlers);
16711Sigor@sysoev.ru 
16877Smax.romanov@nginx.com     nxt_port_read_enable(task, port);
16911Sigor@sysoev.ru }
17011Sigor@sysoev.ru 
17111Sigor@sysoev.ru 
17211Sigor@sysoev.ru static void
nxt_port_handler(nxt_task_t * task,nxt_port_recv_msg_t * msg)17314Sigor@sysoev.ru nxt_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
17411Sigor@sysoev.ru {
17514Sigor@sysoev.ru     nxt_port_handler_t  *handlers;
17611Sigor@sysoev.ru 
177125Smax.romanov@nginx.com     if (nxt_fast_path(msg->port_msg.type < NXT_PORT_MSG_MAX)) {
17811Sigor@sysoev.ru 
1792014Smax.romanov@nginx.com         nxt_debug(task, "port %d: message type:%uD fds:%d,%d",
1802014Smax.romanov@nginx.com                   msg->port->socket.fd, msg->port_msg.type,
1812014Smax.romanov@nginx.com                   msg->fd[0], msg->fd[1]);
18211Sigor@sysoev.ru 
18311Sigor@sysoev.ru         handlers = msg->port->data;
18442Smax.romanov@nginx.com         handlers[msg->port_msg.type](task, msg);
18511Sigor@sysoev.ru 
18611Sigor@sysoev.ru         return;
18711Sigor@sysoev.ru     }
18811Sigor@sysoev.ru 
189564Svbart@nginx.com     nxt_alert(task, "port %d: unknown message type:%uD",
190564Svbart@nginx.com               msg->port->socket.fd, msg->port_msg.type);
19111Sigor@sysoev.ru }
19211Sigor@sysoev.ru 
19311Sigor@sysoev.ru 
19411Sigor@sysoev.ru void
nxt_port_quit_handler(nxt_task_t * task,nxt_port_recv_msg_t * msg)19514Sigor@sysoev.ru nxt_port_quit_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
19611Sigor@sysoev.ru {
197697Sigor@sysoev.ru     nxt_runtime_quit(task, 0);
19811Sigor@sysoev.ru }
19911Sigor@sysoev.ru 
20011Sigor@sysoev.ru 
2011555Smax.romanov@nginx.com /* TODO join with process_ready and move to nxt_main_process.c */
202366Smax.romanov@nginx.com nxt_inline void
nxt_port_send_new_port(nxt_task_t * task,nxt_runtime_t * rt,nxt_port_t * new_port,uint32_t stream)20320Sigor@sysoev.ru nxt_port_send_new_port(nxt_task_t *task, nxt_runtime_t *rt,
204141Smax.romanov@nginx.com     nxt_port_t *new_port, uint32_t stream)
20511Sigor@sysoev.ru {
206141Smax.romanov@nginx.com     nxt_port_t     *port;
20781Smax.romanov@nginx.com     nxt_process_t  *process;
20811Sigor@sysoev.ru 
209141Smax.romanov@nginx.com     nxt_debug(task, "new port %d for process %PI",
210141Smax.romanov@nginx.com               new_port->pair[1], new_port->pid);
21111Sigor@sysoev.ru 
212277Sigor@sysoev.ru     nxt_runtime_process_each(rt, process) {
213277Sigor@sysoev.ru 
21442Smax.romanov@nginx.com         if (process->pid == new_port->pid || process->pid == nxt_pid) {
21511Sigor@sysoev.ru             continue;
21611Sigor@sysoev.ru         }
21711Sigor@sysoev.ru 
218141Smax.romanov@nginx.com         port = nxt_process_port_first(process);
219141Smax.romanov@nginx.com 
220*2031SN/A         if (nxt_proc_send_matrix[port->type][new_port->type]) {
221141Smax.romanov@nginx.com             (void) nxt_port_send_port(task, port, new_port, stream);
222141Smax.romanov@nginx.com         }
223277Sigor@sysoev.ru 
224277Sigor@sysoev.ru     } nxt_runtime_process_loop;
22511Sigor@sysoev.ru }
22611Sigor@sysoev.ru 
22711Sigor@sysoev.ru 
22881Smax.romanov@nginx.com nxt_int_t
nxt_port_send_port(nxt_task_t * task,nxt_port_t * port,nxt_port_t * new_port,uint32_t stream)229141Smax.romanov@nginx.com nxt_port_send_port(nxt_task_t *task, nxt_port_t *port, nxt_port_t *new_port,
230141Smax.romanov@nginx.com     uint32_t stream)
23181Smax.romanov@nginx.com {
23281Smax.romanov@nginx.com     nxt_buf_t                *b;
23381Smax.romanov@nginx.com     nxt_port_msg_new_port_t  *msg;
23481Smax.romanov@nginx.com 
235342Smax.romanov@nginx.com     b = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool,
236342Smax.romanov@nginx.com                              sizeof(nxt_port_data_t));
23781Smax.romanov@nginx.com     if (nxt_slow_path(b == NULL)) {
23881Smax.romanov@nginx.com         return NXT_ERROR;
23981Smax.romanov@nginx.com     }
24081Smax.romanov@nginx.com 
24181Smax.romanov@nginx.com     nxt_debug(task, "send port %FD to process %PI",
24281Smax.romanov@nginx.com               new_port->pair[1], port->pid);
24381Smax.romanov@nginx.com 
24481Smax.romanov@nginx.com     b->mem.free += sizeof(nxt_port_msg_new_port_t);
24581Smax.romanov@nginx.com     msg = (nxt_port_msg_new_port_t *) b->mem.pos;
24681Smax.romanov@nginx.com 
24781Smax.romanov@nginx.com     msg->id = new_port->id;
24881Smax.romanov@nginx.com     msg->pid = new_port->pid;
24981Smax.romanov@nginx.com     msg->max_size = port->max_size;
25081Smax.romanov@nginx.com     msg->max_share = port->max_share;
25181Smax.romanov@nginx.com     msg->type = new_port->type;
25281Smax.romanov@nginx.com 
2531555Smax.romanov@nginx.com     return nxt_port_socket_write2(task, port, NXT_PORT_MSG_NEW_PORT,
2541555Smax.romanov@nginx.com                                   new_port->pair[1], new_port->queue_fd,
2551555Smax.romanov@nginx.com                                   stream, 0, b);
25681Smax.romanov@nginx.com }
25781Smax.romanov@nginx.com 
25881Smax.romanov@nginx.com 
25911Sigor@sysoev.ru void
nxt_port_new_port_handler(nxt_task_t * task,nxt_port_recv_msg_t * msg)26014Sigor@sysoev.ru nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
26111Sigor@sysoev.ru {
26211Sigor@sysoev.ru     nxt_port_t               *port;
26320Sigor@sysoev.ru     nxt_runtime_t            *rt;
26414Sigor@sysoev.ru     nxt_port_msg_new_port_t  *new_port_msg;
26511Sigor@sysoev.ru 
26620Sigor@sysoev.ru     rt = task->thread->runtime;
26711Sigor@sysoev.ru 
26842Smax.romanov@nginx.com     new_port_msg = (nxt_port_msg_new_port_t *) msg->buf->mem.pos;
26942Smax.romanov@nginx.com 
270352Smax.romanov@nginx.com     /* TODO check b size and make plain */
271352Smax.romanov@nginx.com 
272141Smax.romanov@nginx.com     nxt_debug(task, "new port %d received for process %PI:%d",
2731558Smax.romanov@nginx.com               msg->fd[0], new_port_msg->pid, new_port_msg->id);
274141Smax.romanov@nginx.com 
275141Smax.romanov@nginx.com     port = nxt_runtime_port_find(rt, new_port_msg->pid, new_port_msg->id);
276141Smax.romanov@nginx.com     if (port != NULL) {
277141Smax.romanov@nginx.com         nxt_debug(task, "port %PI:%d already exists", new_port_msg->pid,
278141Smax.romanov@nginx.com               new_port_msg->id);
279141Smax.romanov@nginx.com 
2801998St.nateldemoura@f5.com         msg->u.new_port = port;
2811998St.nateldemoura@f5.com 
2821558Smax.romanov@nginx.com         nxt_fd_close(msg->fd[0]);
2831558Smax.romanov@nginx.com         msg->fd[0] = -1;
284141Smax.romanov@nginx.com         return;
285141Smax.romanov@nginx.com     }
286141Smax.romanov@nginx.com 
2871254Shongzhidao@gmail.com     port = nxt_runtime_process_port_create(task, rt, new_port_msg->pid,
2881254Shongzhidao@gmail.com                                            new_port_msg->id,
2891254Shongzhidao@gmail.com                                            new_port_msg->type);
2901254Shongzhidao@gmail.com     if (nxt_slow_path(port == NULL)) {
29120Sigor@sysoev.ru         return;
29220Sigor@sysoev.ru     }
29320Sigor@sysoev.ru 
2941558Smax.romanov@nginx.com     nxt_fd_nonblocking(task, msg->fd[0]);
295799Smax.romanov@nginx.com 
29614Sigor@sysoev.ru     port->pair[0] = -1;
2971558Smax.romanov@nginx.com     port->pair[1] = msg->fd[0];
29814Sigor@sysoev.ru     port->max_size = new_port_msg->max_size;
29914Sigor@sysoev.ru     port->max_share = new_port_msg->max_share;
30011Sigor@sysoev.ru 
30114Sigor@sysoev.ru     port->socket.task = task;
30214Sigor@sysoev.ru 
30311Sigor@sysoev.ru     nxt_port_write_enable(task, port);
304141Smax.romanov@nginx.com 
305347Smax.romanov@nginx.com     msg->u.new_port = port;
306141Smax.romanov@nginx.com }
307141Smax.romanov@nginx.com 
3081555Smax.romanov@nginx.com /* TODO move to nxt_main_process.c */
309141Smax.romanov@nginx.com void
nxt_port_process_ready_handler(nxt_task_t * task,nxt_port_recv_msg_t * msg)310320Smax.romanov@nginx.com nxt_port_process_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
311141Smax.romanov@nginx.com {
312141Smax.romanov@nginx.com     nxt_port_t     *port;
313141Smax.romanov@nginx.com     nxt_process_t  *process;
314141Smax.romanov@nginx.com     nxt_runtime_t  *rt;
315141Smax.romanov@nginx.com 
316141Smax.romanov@nginx.com     rt = task->thread->runtime;
317141Smax.romanov@nginx.com 
318349Smax.romanov@nginx.com     process = nxt_runtime_process_find(rt, msg->port_msg.pid);
319141Smax.romanov@nginx.com     if (nxt_slow_path(process == NULL)) {
320141Smax.romanov@nginx.com         return;
321141Smax.romanov@nginx.com     }
322141Smax.romanov@nginx.com 
3231488St.nateldemoura@f5.com     nxt_assert(process->state != NXT_PROCESS_STATE_READY);
3241488St.nateldemoura@f5.com 
3251488St.nateldemoura@f5.com     process->state = NXT_PROCESS_STATE_READY;
326141Smax.romanov@nginx.com 
327551Smax.romanov@nginx.com     nxt_assert(!nxt_queue_is_empty(&process->ports));
328382Smax.romanov@nginx.com 
329141Smax.romanov@nginx.com     port = nxt_process_port_first(process);
330141Smax.romanov@nginx.com 
331141Smax.romanov@nginx.com     nxt_debug(task, "process %PI ready", msg->port_msg.pid);
332141Smax.romanov@nginx.com 
3331558Smax.romanov@nginx.com     if (msg->fd[0] != -1) {
3341558Smax.romanov@nginx.com         port->queue_fd = msg->fd[0];
3351555Smax.romanov@nginx.com         port->queue = nxt_mem_mmap(NULL, sizeof(nxt_port_queue_t),
3361558Smax.romanov@nginx.com                                    PROT_READ | PROT_WRITE, MAP_SHARED,
3371558Smax.romanov@nginx.com                                    msg->fd[0], 0);
3381555Smax.romanov@nginx.com     }
3391555Smax.romanov@nginx.com 
340196Smax.romanov@nginx.com     nxt_port_send_new_port(task, rt, port, msg->port_msg.stream);
34111Sigor@sysoev.ru }
34211Sigor@sysoev.ru 
34311Sigor@sysoev.ru 
34411Sigor@sysoev.ru void
nxt_port_mmap_handler(nxt_task_t * task,nxt_port_recv_msg_t * msg)34542Smax.romanov@nginx.com nxt_port_mmap_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
34642Smax.romanov@nginx.com {
34742Smax.romanov@nginx.com     nxt_runtime_t  *rt;
34842Smax.romanov@nginx.com     nxt_process_t  *process;
34942Smax.romanov@nginx.com 
35042Smax.romanov@nginx.com     rt = task->thread->runtime;
35142Smax.romanov@nginx.com 
3521558Smax.romanov@nginx.com     if (nxt_slow_path(msg->fd[0] == -1)) {
35342Smax.romanov@nginx.com         nxt_log(task, NXT_LOG_WARN, "invalid fd passed with mmap message");
35442Smax.romanov@nginx.com 
35542Smax.romanov@nginx.com         return;
35642Smax.romanov@nginx.com     }
35742Smax.romanov@nginx.com 
358196Smax.romanov@nginx.com     process = nxt_runtime_process_find(rt, msg->port_msg.pid);
35942Smax.romanov@nginx.com     if (nxt_slow_path(process == NULL)) {
36042Smax.romanov@nginx.com         nxt_log(task, NXT_LOG_WARN, "failed to get process #%PI",
36142Smax.romanov@nginx.com                 msg->port_msg.pid);
36242Smax.romanov@nginx.com 
36342Smax.romanov@nginx.com         goto fail_close;
36442Smax.romanov@nginx.com     }
36542Smax.romanov@nginx.com 
3661558Smax.romanov@nginx.com     nxt_port_incoming_port_mmap(task, process, msg->fd[0]);
36742Smax.romanov@nginx.com 
36842Smax.romanov@nginx.com fail_close:
36942Smax.romanov@nginx.com 
3701558Smax.romanov@nginx.com     nxt_fd_close(msg->fd[0]);
37142Smax.romanov@nginx.com }
37242Smax.romanov@nginx.com 
37342Smax.romanov@nginx.com 
37442Smax.romanov@nginx.com void
nxt_port_change_log_file(nxt_task_t * task,nxt_runtime_t * rt,nxt_uint_t slot,nxt_fd_t fd)37520Sigor@sysoev.ru nxt_port_change_log_file(nxt_task_t *task, nxt_runtime_t *rt, nxt_uint_t slot,
37620Sigor@sysoev.ru     nxt_fd_t fd)
37711Sigor@sysoev.ru {
37820Sigor@sysoev.ru     nxt_buf_t      *b;
37920Sigor@sysoev.ru     nxt_port_t     *port;
38020Sigor@sysoev.ru     nxt_process_t  *process;
38111Sigor@sysoev.ru 
38214Sigor@sysoev.ru     nxt_debug(task, "change log file #%ui fd:%FD", slot, fd);
38311Sigor@sysoev.ru 
384277Sigor@sysoev.ru     nxt_runtime_process_each(rt, process) {
385277Sigor@sysoev.ru 
38642Smax.romanov@nginx.com         if (nxt_pid == process->pid) {
38742Smax.romanov@nginx.com             continue;
38842Smax.romanov@nginx.com         }
38911Sigor@sysoev.ru 
39042Smax.romanov@nginx.com         port = nxt_process_port_first(process);
39111Sigor@sysoev.ru 
3921998St.nateldemoura@f5.com         b = nxt_buf_mem_alloc(task->thread->engine->mem_pool,
3931998St.nateldemoura@f5.com                               sizeof(nxt_uint_t), 0);
39411Sigor@sysoev.ru         if (nxt_slow_path(b == NULL)) {
39511Sigor@sysoev.ru             continue;
39611Sigor@sysoev.ru         }
39711Sigor@sysoev.ru 
3981998St.nateldemoura@f5.com         b->mem.free = nxt_cpymem(b->mem.free, &slot, sizeof(nxt_uint_t));
39911Sigor@sysoev.ru 
40020Sigor@sysoev.ru         (void) nxt_port_socket_write(task, port, NXT_PORT_MSG_CHANGE_FILE,
40142Smax.romanov@nginx.com                                      fd, 0, 0, b);
402277Sigor@sysoev.ru 
403277Sigor@sysoev.ru     } nxt_runtime_process_loop;
40411Sigor@sysoev.ru }
40511Sigor@sysoev.ru 
40611Sigor@sysoev.ru 
40711Sigor@sysoev.ru void
nxt_port_change_log_file_handler(nxt_task_t * task,nxt_port_recv_msg_t * msg)40814Sigor@sysoev.ru nxt_port_change_log_file_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
40911Sigor@sysoev.ru {
41020Sigor@sysoev.ru     nxt_buf_t      *b;
41120Sigor@sysoev.ru     nxt_uint_t     slot;
41220Sigor@sysoev.ru     nxt_file_t     *log_file;
41320Sigor@sysoev.ru     nxt_runtime_t  *rt;
41411Sigor@sysoev.ru 
41520Sigor@sysoev.ru     rt = task->thread->runtime;
41611Sigor@sysoev.ru 
41711Sigor@sysoev.ru     b = msg->buf;
41811Sigor@sysoev.ru     slot = *(nxt_uint_t *) b->mem.pos;
41911Sigor@sysoev.ru 
42020Sigor@sysoev.ru     log_file = nxt_list_elt(rt->log_files, slot);
42111Sigor@sysoev.ru 
4221558Smax.romanov@nginx.com     nxt_debug(task, "change log file %FD:%FD", msg->fd[0], log_file->fd);
42311Sigor@sysoev.ru 
42411Sigor@sysoev.ru     /*
42511Sigor@sysoev.ru      * The old log file descriptor must be closed at the moment when no
42611Sigor@sysoev.ru      * other threads use it.  dup2() allows to use the old file descriptor
42711Sigor@sysoev.ru      * for new log file.  This change is performed atomically in the kernel.
42811Sigor@sysoev.ru      */
4291558Smax.romanov@nginx.com     if (nxt_file_redirect(log_file, msg->fd[0]) == NXT_OK) {
43011Sigor@sysoev.ru         if (slot == 0) {
43111Sigor@sysoev.ru             (void) nxt_file_stderr(log_file);
43211Sigor@sysoev.ru         }
43311Sigor@sysoev.ru     }
43411Sigor@sysoev.ru }
43511Sigor@sysoev.ru 
43611Sigor@sysoev.ru 
43711Sigor@sysoev.ru void
nxt_port_data_handler(nxt_task_t * task,nxt_port_recv_msg_t * msg)43814Sigor@sysoev.ru nxt_port_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
43911Sigor@sysoev.ru {
44042Smax.romanov@nginx.com     size_t     dump_size;
44111Sigor@sysoev.ru     nxt_buf_t  *b;
44211Sigor@sysoev.ru 
44311Sigor@sysoev.ru     b = msg->buf;
44442Smax.romanov@nginx.com     dump_size = b->mem.free - b->mem.pos;
44511Sigor@sysoev.ru 
44642Smax.romanov@nginx.com     if (dump_size > 300) {
44742Smax.romanov@nginx.com         dump_size = 300;
44842Smax.romanov@nginx.com     }
44942Smax.romanov@nginx.com 
45042Smax.romanov@nginx.com     nxt_debug(task, "data: %*s", dump_size, b->mem.pos);
45111Sigor@sysoev.ru }
45211Sigor@sysoev.ru 
45311Sigor@sysoev.ru 
45411Sigor@sysoev.ru void
nxt_port_remove_notify_others(nxt_task_t * task,nxt_process_t * process)4551997St.nateldemoura@f5.com nxt_port_remove_notify_others(nxt_task_t *task, nxt_process_t *process)
4561997St.nateldemoura@f5.com {
4571997St.nateldemoura@f5.com     nxt_pid_t           pid;
4581997St.nateldemoura@f5.com     nxt_buf_t           *buf;
4591997St.nateldemoura@f5.com     nxt_port_t          *port;
4601997St.nateldemoura@f5.com     nxt_runtime_t       *rt;
4611997St.nateldemoura@f5.com     nxt_process_t       *p;
4621997St.nateldemoura@f5.com     nxt_process_type_t  ptype;
4631997St.nateldemoura@f5.com 
4641997St.nateldemoura@f5.com     pid = process->pid;
4651997St.nateldemoura@f5.com 
4661997St.nateldemoura@f5.com     ptype = nxt_process_type(process);
4671997St.nateldemoura@f5.com 
4681997St.nateldemoura@f5.com     rt = task->thread->runtime;
4691997St.nateldemoura@f5.com 
4701997St.nateldemoura@f5.com     nxt_runtime_process_each(rt, p) {
4711997St.nateldemoura@f5.com 
4721997St.nateldemoura@f5.com         if (p->pid == nxt_pid
4731997St.nateldemoura@f5.com             || p->pid == pid
4741997St.nateldemoura@f5.com             || nxt_queue_is_empty(&p->ports))
4751997St.nateldemoura@f5.com         {
4761997St.nateldemoura@f5.com             continue;
4771997St.nateldemoura@f5.com         }
4781997St.nateldemoura@f5.com 
4791997St.nateldemoura@f5.com         port = nxt_process_port_first(p);
4801997St.nateldemoura@f5.com 
4811997St.nateldemoura@f5.com         if (nxt_proc_remove_notify_matrix[ptype][port->type] == 0) {
4821997St.nateldemoura@f5.com             continue;
4831997St.nateldemoura@f5.com         }
4841997St.nateldemoura@f5.com 
4851997St.nateldemoura@f5.com         buf = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool,
4861997St.nateldemoura@f5.com                                    sizeof(pid));
4871997St.nateldemoura@f5.com 
4881997St.nateldemoura@f5.com         if (nxt_slow_path(buf == NULL)) {
4891997St.nateldemoura@f5.com             continue;
4901997St.nateldemoura@f5.com         }
4911997St.nateldemoura@f5.com 
4921997St.nateldemoura@f5.com         buf->mem.free = nxt_cpymem(buf->mem.free, &pid, sizeof(pid));
4931997St.nateldemoura@f5.com 
4941997St.nateldemoura@f5.com         nxt_port_socket_write(task, port, NXT_PORT_MSG_REMOVE_PID, -1,
4951997St.nateldemoura@f5.com                               process->stream, 0, buf);
4961997St.nateldemoura@f5.com 
4971997St.nateldemoura@f5.com     } nxt_runtime_process_loop;
4981997St.nateldemoura@f5.com }
4991997St.nateldemoura@f5.com 
5001997St.nateldemoura@f5.com 
5011997St.nateldemoura@f5.com void
nxt_port_remove_pid_handler(nxt_task_t * task,nxt_port_recv_msg_t * msg)502125Smax.romanov@nginx.com nxt_port_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
503125Smax.romanov@nginx.com {
5041998St.nateldemoura@f5.com     nxt_pid_t  pid;
5051998St.nateldemoura@f5.com     nxt_buf_t  *buf;
506125Smax.romanov@nginx.com 
507192Smax.romanov@nginx.com     buf = msg->buf;
508192Smax.romanov@nginx.com 
509192Smax.romanov@nginx.com     nxt_assert(nxt_buf_used_size(buf) == sizeof(pid));
510192Smax.romanov@nginx.com 
5111998St.nateldemoura@f5.com     nxt_memcpy(&pid, buf->mem.pos, sizeof(nxt_pid_t));
5121998St.nateldemoura@f5.com 
5131998St.nateldemoura@f5.com     nxt_port_remove_pid(task, msg, pid);
5141998St.nateldemoura@f5.com }
5151998St.nateldemoura@f5.com 
5161998St.nateldemoura@f5.com 
5171998St.nateldemoura@f5.com static void
nxt_port_remove_pid(nxt_task_t * task,nxt_port_recv_msg_t * msg,nxt_pid_t pid)5181998St.nateldemoura@f5.com nxt_port_remove_pid(nxt_task_t *task, nxt_port_recv_msg_t *msg,
5191998St.nateldemoura@f5.com     nxt_pid_t pid)
5201998St.nateldemoura@f5.com {
5211998St.nateldemoura@f5.com     nxt_runtime_t  *rt;
5221998St.nateldemoura@f5.com     nxt_process_t  *process;
523192Smax.romanov@nginx.com 
524347Smax.romanov@nginx.com     msg->u.removed_pid = pid;
525347Smax.romanov@nginx.com 
526196Smax.romanov@nginx.com     nxt_debug(task, "port remove pid %PI handler", pid);
527196Smax.romanov@nginx.com 
528125Smax.romanov@nginx.com     rt = task->thread->runtime;
529125Smax.romanov@nginx.com 
530190Smax.romanov@nginx.com     nxt_port_rpc_remove_peer(task, msg->port, pid);
531190Smax.romanov@nginx.com 
532125Smax.romanov@nginx.com     process = nxt_runtime_process_find(rt, pid);
533125Smax.romanov@nginx.com 
534125Smax.romanov@nginx.com     if (process) {
535349Smax.romanov@nginx.com         nxt_process_close_ports(task, process);
536125Smax.romanov@nginx.com     }
537125Smax.romanov@nginx.com }
538125Smax.romanov@nginx.com 
539125Smax.romanov@nginx.com 
540125Smax.romanov@nginx.com void
nxt_port_empty_handler(nxt_task_t * task,nxt_port_recv_msg_t * msg)54114Sigor@sysoev.ru nxt_port_empty_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
54211Sigor@sysoev.ru {
54311Sigor@sysoev.ru     nxt_debug(task, "port empty handler");
54411Sigor@sysoev.ru }
545343Smax.romanov@nginx.com 
546343Smax.romanov@nginx.com 
547343Smax.romanov@nginx.com typedef struct {
548343Smax.romanov@nginx.com     nxt_work_t               work;
549343Smax.romanov@nginx.com     nxt_port_t               *port;
550343Smax.romanov@nginx.com     nxt_port_post_handler_t  handler;
551343Smax.romanov@nginx.com } nxt_port_work_t;
552343Smax.romanov@nginx.com 
553343Smax.romanov@nginx.com 
554343Smax.romanov@nginx.com static void
nxt_port_post_handler(nxt_task_t * task,void * obj,void * data)555343Smax.romanov@nginx.com nxt_port_post_handler(nxt_task_t *task, void *obj, void *data)
556343Smax.romanov@nginx.com {
557343Smax.romanov@nginx.com     nxt_port_t               *port;
558343Smax.romanov@nginx.com     nxt_port_work_t          *pw;
559343Smax.romanov@nginx.com     nxt_port_post_handler_t  handler;
560343Smax.romanov@nginx.com 
561343Smax.romanov@nginx.com     pw = obj;
562343Smax.romanov@nginx.com     port = pw->port;
563343Smax.romanov@nginx.com     handler = pw->handler;
564343Smax.romanov@nginx.com 
565343Smax.romanov@nginx.com     nxt_free(pw);
566343Smax.romanov@nginx.com 
567343Smax.romanov@nginx.com     handler(task, port, data);
568343Smax.romanov@nginx.com 
569343Smax.romanov@nginx.com     nxt_port_use(task, port, -1);
570343Smax.romanov@nginx.com }
571343Smax.romanov@nginx.com 
572343Smax.romanov@nginx.com 
573343Smax.romanov@nginx.com nxt_int_t
nxt_port_post(nxt_task_t * task,nxt_port_t * port,nxt_port_post_handler_t handler,void * data)574343Smax.romanov@nginx.com nxt_port_post(nxt_task_t *task, nxt_port_t *port,
575343Smax.romanov@nginx.com     nxt_port_post_handler_t handler, void *data)
576343Smax.romanov@nginx.com {
577343Smax.romanov@nginx.com     nxt_port_work_t  *pw;
578343Smax.romanov@nginx.com 
579343Smax.romanov@nginx.com     if (task->thread->engine == port->engine) {
580343Smax.romanov@nginx.com         handler(task, port, data);
581343Smax.romanov@nginx.com 
582343Smax.romanov@nginx.com         return NXT_OK;
583343Smax.romanov@nginx.com     }
584343Smax.romanov@nginx.com 
585343Smax.romanov@nginx.com     pw = nxt_zalloc(sizeof(nxt_port_work_t));
586343Smax.romanov@nginx.com 
587343Smax.romanov@nginx.com     if (nxt_slow_path(pw == NULL)) {
5881008Szelenkov@nginx.com         return NXT_ERROR;
589343Smax.romanov@nginx.com     }
590343Smax.romanov@nginx.com 
591343Smax.romanov@nginx.com     nxt_atomic_fetch_add(&port->use_count, 1);
592343Smax.romanov@nginx.com 
593343Smax.romanov@nginx.com     pw->work.handler = nxt_port_post_handler;
594343Smax.romanov@nginx.com     pw->work.task = &port->engine->task;
595343Smax.romanov@nginx.com     pw->work.obj = pw;
596343Smax.romanov@nginx.com     pw->work.data = data;
597343Smax.romanov@nginx.com 
598343Smax.romanov@nginx.com     pw->port = port;
599343Smax.romanov@nginx.com     pw->handler = handler;
600343Smax.romanov@nginx.com 
601343Smax.romanov@nginx.com     nxt_event_engine_post(port->engine, &pw->work);
602343Smax.romanov@nginx.com 
603343Smax.romanov@nginx.com     return NXT_OK;
604343Smax.romanov@nginx.com }
605343Smax.romanov@nginx.com 
606343Smax.romanov@nginx.com 
607343Smax.romanov@nginx.com static void
nxt_port_release_handler(nxt_task_t * task,nxt_port_t * port,void * data)608343Smax.romanov@nginx.com nxt_port_release_handler(nxt_task_t *task, nxt_port_t *port, void *data)
609343Smax.romanov@nginx.com {
610343Smax.romanov@nginx.com     /* no op */
611343Smax.romanov@nginx.com }
612343Smax.romanov@nginx.com 
613343Smax.romanov@nginx.com 
614343Smax.romanov@nginx.com void
nxt_port_use(nxt_task_t * task,nxt_port_t * port,int i)615343Smax.romanov@nginx.com nxt_port_use(nxt_task_t *task, nxt_port_t *port, int i)
616343Smax.romanov@nginx.com {
617343Smax.romanov@nginx.com     int  c;
618343Smax.romanov@nginx.com 
619343Smax.romanov@nginx.com     c = nxt_atomic_fetch_add(&port->use_count, i);
620343Smax.romanov@nginx.com 
621343Smax.romanov@nginx.com     if (i < 0 && c == -i) {
622343Smax.romanov@nginx.com 
6231180Smax.romanov@nginx.com         if (port->engine == NULL || task->thread->engine == port->engine) {
624343Smax.romanov@nginx.com             nxt_port_release(task, port);
625343Smax.romanov@nginx.com 
626343Smax.romanov@nginx.com             return;
627343Smax.romanov@nginx.com         }
628343Smax.romanov@nginx.com 
629343Smax.romanov@nginx.com         nxt_port_post(task, port, nxt_port_release_handler, NULL);
630343Smax.romanov@nginx.com     }
631343Smax.romanov@nginx.com }
632