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