xref: /unit/src/nxt_port.c (revision 1998:c8790d2a89bb)
1 
2 /*
3  * Copyright (C) Igor Sysoev
4  * Copyright (C) NGINX, Inc.
5  */
6 
7 #include <nxt_main.h>
8 #include <nxt_runtime.h>
9 #include <nxt_port.h>
10 #include <nxt_router.h>
11 #include <nxt_app_queue.h>
12 #include <nxt_port_queue.h>
13 
14 
15 static void nxt_port_remove_pid(nxt_task_t *task, nxt_port_recv_msg_t *msg,
16     nxt_pid_t pid);
17 static void nxt_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
18 
19 static nxt_atomic_uint_t nxt_port_last_id = 1;
20 
21 
22 static void
23 nxt_port_mp_cleanup(nxt_task_t *task, void *obj, void *data)
24 {
25     nxt_mp_t    *mp;
26     nxt_port_t  *port;
27 
28     port = obj;
29     mp = data;
30 
31     nxt_assert(port->pair[0] == -1);
32     nxt_assert(port->pair[1] == -1);
33 
34     nxt_assert(port->use_count == 0);
35     nxt_assert(port->app_link.next == NULL);
36     nxt_assert(port->idle_link.next == NULL);
37 
38     nxt_assert(nxt_queue_is_empty(&port->messages));
39     nxt_assert(nxt_lvlhsh_is_empty(&port->rpc_streams));
40     nxt_assert(nxt_lvlhsh_is_empty(&port->rpc_peers));
41 
42     nxt_thread_mutex_destroy(&port->write_mutex);
43 
44     nxt_mp_free(mp, port);
45 }
46 
47 
48 nxt_port_t *
49 nxt_port_new(nxt_task_t *task, nxt_port_id_t id, nxt_pid_t pid,
50     nxt_process_type_t type)
51 {
52     nxt_mp_t    *mp;
53     nxt_port_t  *port;
54 
55     mp = nxt_mp_create(1024, 128, 256, 32);
56 
57     if (nxt_slow_path(mp == NULL)) {
58         return NULL;
59     }
60 
61     port = nxt_mp_zalloc(mp, sizeof(nxt_port_t));
62 
63     if (nxt_fast_path(port != NULL)) {
64         port->id = id;
65         port->pid = pid;
66         port->type = type;
67         port->mem_pool = mp;
68         port->use_count = 1;
69 
70         nxt_mp_cleanup(mp, nxt_port_mp_cleanup, task, port, mp);
71 
72         nxt_queue_init(&port->messages);
73         nxt_thread_mutex_create(&port->write_mutex);
74 
75         port->queue_fd = -1;
76 
77     } else {
78         nxt_mp_destroy(mp);
79     }
80 
81     nxt_thread_log_debug("port %p %d:%d new, type %d", port, pid, id, type);
82 
83     return port;
84 }
85 
86 
87 void
88 nxt_port_close(nxt_task_t *task, nxt_port_t *port)
89 {
90     size_t  size;
91 
92     nxt_debug(task, "port %p %d:%d close, type %d", port, port->pid,
93               port->id, port->type);
94 
95     if (port->pair[0] != -1) {
96         nxt_port_rpc_close(task, port);
97 
98         nxt_fd_close(port->pair[0]);
99         port->pair[0] = -1;
100     }
101 
102     if (port->pair[1] != -1) {
103         nxt_fd_close(port->pair[1]);
104         port->pair[1] = -1;
105 
106         if (port->app != NULL) {
107             nxt_router_app_port_close(task, port);
108         }
109     }
110 
111     if (port->queue_fd != -1) {
112         nxt_fd_close(port->queue_fd);
113         port->queue_fd = -1;
114     }
115 
116     if (port->queue != NULL) {
117         size = (port->id == (nxt_port_id_t) -1) ? sizeof(nxt_app_queue_t)
118                                                 : sizeof(nxt_port_queue_t);
119         nxt_mem_munmap(port->queue, size);
120 
121         port->queue = NULL;
122     }
123 }
124 
125 
126 static void
127 nxt_port_release(nxt_task_t *task, nxt_port_t *port)
128 {
129     nxt_debug(task, "port %p %d:%d release, type %d", port, port->pid,
130               port->id, port->type);
131 
132     port->app = NULL;
133 
134     if (port->link.next != NULL) {
135         nxt_assert(port->process != NULL);
136 
137         nxt_process_port_remove(port);
138 
139         nxt_process_use(task, port->process, -1);
140     }
141 
142     nxt_mp_release(port->mem_pool);
143 }
144 
145 
146 nxt_port_id_t
147 nxt_port_get_next_id()
148 {
149     return nxt_atomic_fetch_add(&nxt_port_last_id, 1);
150 }
151 
152 
153 void
154 nxt_port_reset_next_id()
155 {
156     nxt_port_last_id = 1;
157 }
158 
159 
160 void
161 nxt_port_enable(nxt_task_t *task, nxt_port_t *port,
162     const nxt_port_handlers_t *handlers)
163 {
164     port->pid = nxt_pid;
165     port->handler = nxt_port_handler;
166     port->data = (nxt_port_handler_t *) (handlers);
167 
168     nxt_port_read_enable(task, port);
169 }
170 
171 
172 static void
173 nxt_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
174 {
175     nxt_port_handler_t  *handlers;
176 
177     if (nxt_fast_path(msg->port_msg.type < NXT_PORT_MSG_MAX)) {
178 
179         nxt_debug(task, "port %d: message type:%uD",
180                   msg->port->socket.fd, msg->port_msg.type);
181 
182         handlers = msg->port->data;
183         handlers[msg->port_msg.type](task, msg);
184 
185         return;
186     }
187 
188     nxt_alert(task, "port %d: unknown message type:%uD",
189               msg->port->socket.fd, msg->port_msg.type);
190 }
191 
192 
193 void
194 nxt_port_quit_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
195 {
196     nxt_runtime_quit(task, 0);
197 }
198 
199 
200 /* TODO join with process_ready and move to nxt_main_process.c */
201 nxt_inline void
202 nxt_port_send_new_port(nxt_task_t *task, nxt_runtime_t *rt,
203     nxt_port_t *new_port, uint32_t stream)
204 {
205     nxt_port_t     *port;
206     nxt_process_t  *process;
207 
208     nxt_debug(task, "new port %d for process %PI",
209               new_port->pair[1], new_port->pid);
210 
211     nxt_runtime_process_each(rt, process) {
212 
213         if (process->pid == new_port->pid || process->pid == nxt_pid) {
214             continue;
215         }
216 
217         port = nxt_process_port_first(process);
218 
219         if (nxt_proc_conn_matrix[port->type][new_port->type]) {
220             (void) nxt_port_send_port(task, port, new_port, stream);
221         }
222 
223     } nxt_runtime_process_loop;
224 }
225 
226 
227 nxt_int_t
228 nxt_port_send_port(nxt_task_t *task, nxt_port_t *port, nxt_port_t *new_port,
229     uint32_t stream)
230 {
231     nxt_buf_t                *b;
232     nxt_port_msg_new_port_t  *msg;
233 
234     b = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool,
235                              sizeof(nxt_port_data_t));
236     if (nxt_slow_path(b == NULL)) {
237         return NXT_ERROR;
238     }
239 
240     nxt_debug(task, "send port %FD to process %PI",
241               new_port->pair[1], port->pid);
242 
243     b->mem.free += sizeof(nxt_port_msg_new_port_t);
244     msg = (nxt_port_msg_new_port_t *) b->mem.pos;
245 
246     msg->id = new_port->id;
247     msg->pid = new_port->pid;
248     msg->max_size = port->max_size;
249     msg->max_share = port->max_share;
250     msg->type = new_port->type;
251 
252     return nxt_port_socket_write2(task, port, NXT_PORT_MSG_NEW_PORT,
253                                   new_port->pair[1], new_port->queue_fd,
254                                   stream, 0, b);
255 }
256 
257 
258 void
259 nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
260 {
261     nxt_port_t               *port;
262     nxt_runtime_t            *rt;
263     nxt_port_msg_new_port_t  *new_port_msg;
264 
265     rt = task->thread->runtime;
266 
267     new_port_msg = (nxt_port_msg_new_port_t *) msg->buf->mem.pos;
268 
269     /* TODO check b size and make plain */
270 
271     nxt_debug(task, "new port %d received for process %PI:%d",
272               msg->fd[0], new_port_msg->pid, new_port_msg->id);
273 
274     port = nxt_runtime_port_find(rt, new_port_msg->pid, new_port_msg->id);
275     if (port != NULL) {
276         nxt_debug(task, "port %PI:%d already exists", new_port_msg->pid,
277               new_port_msg->id);
278 
279         msg->u.new_port = port;
280 
281         nxt_fd_close(msg->fd[0]);
282         msg->fd[0] = -1;
283         return;
284     }
285 
286     port = nxt_runtime_process_port_create(task, rt, new_port_msg->pid,
287                                            new_port_msg->id,
288                                            new_port_msg->type);
289     if (nxt_slow_path(port == NULL)) {
290         return;
291     }
292 
293     nxt_fd_nonblocking(task, msg->fd[0]);
294 
295     port->pair[0] = -1;
296     port->pair[1] = msg->fd[0];
297     port->max_size = new_port_msg->max_size;
298     port->max_share = new_port_msg->max_share;
299 
300     port->socket.task = task;
301 
302     nxt_port_write_enable(task, port);
303 
304     msg->u.new_port = port;
305 }
306 
307 /* TODO move to nxt_main_process.c */
308 void
309 nxt_port_process_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
310 {
311     nxt_port_t     *port;
312     nxt_process_t  *process;
313     nxt_runtime_t  *rt;
314 
315     rt = task->thread->runtime;
316 
317     process = nxt_runtime_process_find(rt, msg->port_msg.pid);
318     if (nxt_slow_path(process == NULL)) {
319         return;
320     }
321 
322     nxt_assert(process->state != NXT_PROCESS_STATE_READY);
323 
324     process->state = NXT_PROCESS_STATE_READY;
325 
326     nxt_assert(!nxt_queue_is_empty(&process->ports));
327 
328     port = nxt_process_port_first(process);
329 
330     nxt_debug(task, "process %PI ready", msg->port_msg.pid);
331 
332     if (msg->fd[0] != -1) {
333         port->queue_fd = msg->fd[0];
334         port->queue = nxt_mem_mmap(NULL, sizeof(nxt_port_queue_t),
335                                    PROT_READ | PROT_WRITE, MAP_SHARED,
336                                    msg->fd[0], 0);
337     }
338 
339     nxt_port_send_new_port(task, rt, port, msg->port_msg.stream);
340 }
341 
342 
343 void
344 nxt_port_mmap_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
345 {
346     nxt_runtime_t  *rt;
347     nxt_process_t  *process;
348 
349     rt = task->thread->runtime;
350 
351     if (nxt_slow_path(msg->fd[0] == -1)) {
352         nxt_log(task, NXT_LOG_WARN, "invalid fd passed with mmap message");
353 
354         return;
355     }
356 
357     process = nxt_runtime_process_find(rt, msg->port_msg.pid);
358     if (nxt_slow_path(process == NULL)) {
359         nxt_log(task, NXT_LOG_WARN, "failed to get process #%PI",
360                 msg->port_msg.pid);
361 
362         goto fail_close;
363     }
364 
365     nxt_port_incoming_port_mmap(task, process, msg->fd[0]);
366 
367 fail_close:
368 
369     nxt_fd_close(msg->fd[0]);
370 }
371 
372 
373 void
374 nxt_port_change_log_file(nxt_task_t *task, nxt_runtime_t *rt, nxt_uint_t slot,
375     nxt_fd_t fd)
376 {
377     nxt_buf_t      *b;
378     nxt_port_t     *port;
379     nxt_process_t  *process;
380 
381     nxt_debug(task, "change log file #%ui fd:%FD", slot, fd);
382 
383     nxt_runtime_process_each(rt, process) {
384 
385         if (nxt_pid == process->pid) {
386             continue;
387         }
388 
389         port = nxt_process_port_first(process);
390 
391         b = nxt_buf_mem_alloc(task->thread->engine->mem_pool,
392                               sizeof(nxt_uint_t), 0);
393         if (nxt_slow_path(b == NULL)) {
394             continue;
395         }
396 
397         b->mem.free = nxt_cpymem(b->mem.free, &slot, sizeof(nxt_uint_t));
398 
399         (void) nxt_port_socket_write(task, port, NXT_PORT_MSG_CHANGE_FILE,
400                                      fd, 0, 0, b);
401 
402     } nxt_runtime_process_loop;
403 }
404 
405 
406 void
407 nxt_port_change_log_file_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
408 {
409     nxt_buf_t      *b;
410     nxt_uint_t     slot;
411     nxt_file_t     *log_file;
412     nxt_runtime_t  *rt;
413 
414     rt = task->thread->runtime;
415 
416     b = msg->buf;
417     slot = *(nxt_uint_t *) b->mem.pos;
418 
419     log_file = nxt_list_elt(rt->log_files, slot);
420 
421     nxt_debug(task, "change log file %FD:%FD", msg->fd[0], log_file->fd);
422 
423     /*
424      * The old log file descriptor must be closed at the moment when no
425      * other threads use it.  dup2() allows to use the old file descriptor
426      * for new log file.  This change is performed atomically in the kernel.
427      */
428     if (nxt_file_redirect(log_file, msg->fd[0]) == NXT_OK) {
429         if (slot == 0) {
430             (void) nxt_file_stderr(log_file);
431         }
432     }
433 }
434 
435 
436 void
437 nxt_port_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
438 {
439     size_t     dump_size;
440     nxt_buf_t  *b;
441 
442     b = msg->buf;
443     dump_size = b->mem.free - b->mem.pos;
444 
445     if (dump_size > 300) {
446         dump_size = 300;
447     }
448 
449     nxt_debug(task, "data: %*s", dump_size, b->mem.pos);
450 }
451 
452 
453 void
454 nxt_port_remove_notify_others(nxt_task_t *task, nxt_process_t *process)
455 {
456     nxt_pid_t           pid;
457     nxt_buf_t           *buf;
458     nxt_port_t          *port;
459     nxt_runtime_t       *rt;
460     nxt_process_t       *p;
461     nxt_process_type_t  ptype;
462 
463     pid = process->pid;
464 
465     ptype = nxt_process_type(process);
466 
467     rt = task->thread->runtime;
468 
469     nxt_runtime_process_each(rt, p) {
470 
471         if (p->pid == nxt_pid
472             || p->pid == pid
473             || nxt_queue_is_empty(&p->ports))
474         {
475             continue;
476         }
477 
478         port = nxt_process_port_first(p);
479 
480         if (nxt_proc_remove_notify_matrix[ptype][port->type] == 0) {
481             continue;
482         }
483 
484         buf = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool,
485                                    sizeof(pid));
486 
487         if (nxt_slow_path(buf == NULL)) {
488             continue;
489         }
490 
491         buf->mem.free = nxt_cpymem(buf->mem.free, &pid, sizeof(pid));
492 
493         nxt_port_socket_write(task, port, NXT_PORT_MSG_REMOVE_PID, -1,
494                               process->stream, 0, buf);
495 
496     } nxt_runtime_process_loop;
497 }
498 
499 
500 void
501 nxt_port_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
502 {
503     nxt_pid_t  pid;
504     nxt_buf_t  *buf;
505 
506     buf = msg->buf;
507 
508     nxt_assert(nxt_buf_used_size(buf) == sizeof(pid));
509 
510     nxt_memcpy(&pid, buf->mem.pos, sizeof(nxt_pid_t));
511 
512     nxt_port_remove_pid(task, msg, pid);
513 }
514 
515 
516 static void
517 nxt_port_remove_pid(nxt_task_t *task, nxt_port_recv_msg_t *msg,
518     nxt_pid_t pid)
519 {
520     nxt_runtime_t  *rt;
521     nxt_process_t  *process;
522 
523     msg->u.removed_pid = pid;
524 
525     nxt_debug(task, "port remove pid %PI handler", pid);
526 
527     rt = task->thread->runtime;
528 
529     nxt_port_rpc_remove_peer(task, msg->port, pid);
530 
531     process = nxt_runtime_process_find(rt, pid);
532 
533     if (process) {
534         nxt_process_close_ports(task, process);
535     }
536 }
537 
538 
539 void
540 nxt_port_empty_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
541 {
542     nxt_debug(task, "port empty handler");
543 }
544 
545 
546 typedef struct {
547     nxt_work_t               work;
548     nxt_port_t               *port;
549     nxt_port_post_handler_t  handler;
550 } nxt_port_work_t;
551 
552 
553 static void
554 nxt_port_post_handler(nxt_task_t *task, void *obj, void *data)
555 {
556     nxt_port_t               *port;
557     nxt_port_work_t          *pw;
558     nxt_port_post_handler_t  handler;
559 
560     pw = obj;
561     port = pw->port;
562     handler = pw->handler;
563 
564     nxt_free(pw);
565 
566     handler(task, port, data);
567 
568     nxt_port_use(task, port, -1);
569 }
570 
571 
572 nxt_int_t
573 nxt_port_post(nxt_task_t *task, nxt_port_t *port,
574     nxt_port_post_handler_t handler, void *data)
575 {
576     nxt_port_work_t  *pw;
577 
578     if (task->thread->engine == port->engine) {
579         handler(task, port, data);
580 
581         return NXT_OK;
582     }
583 
584     pw = nxt_zalloc(sizeof(nxt_port_work_t));
585 
586     if (nxt_slow_path(pw == NULL)) {
587         return NXT_ERROR;
588     }
589 
590     nxt_atomic_fetch_add(&port->use_count, 1);
591 
592     pw->work.handler = nxt_port_post_handler;
593     pw->work.task = &port->engine->task;
594     pw->work.obj = pw;
595     pw->work.data = data;
596 
597     pw->port = port;
598     pw->handler = handler;
599 
600     nxt_event_engine_post(port->engine, &pw->work);
601 
602     return NXT_OK;
603 }
604 
605 
606 static void
607 nxt_port_release_handler(nxt_task_t *task, nxt_port_t *port, void *data)
608 {
609     /* no op */
610 }
611 
612 
613 void
614 nxt_port_use(nxt_task_t *task, nxt_port_t *port, int i)
615 {
616     int  c;
617 
618     c = nxt_atomic_fetch_add(&port->use_count, i);
619 
620     if (i < 0 && c == -i) {
621 
622         if (port->engine == NULL || task->thread->engine == port->engine) {
623             nxt_port_release(task, port);
624 
625             return;
626         }
627 
628         nxt_port_post(task, port, nxt_port_release_handler, NULL);
629     }
630 }
631