Processes refactoring.

The cycle has been renamed to the runtime.

1 files changed, 59 insertions, 41 deletions

diff --git a/src/nxt_port.c b/src/nxt_port.c
index 8a56fc6b..1da16587 100644
--- a/src/nxt_port.c
+++ b/src/nxt_port.c
@@ -5,7 +5,7 @@
  */
 
 #include <nxt_main.h>
-#include <nxt_cycle.h>
+#include <nxt_runtime.h>
 #include <nxt_port.h>
 
 
@@ -29,18 +29,26 @@ nxt_port_create(nxt_thread_t *thread, nxt_port_t *port,
 
 
 void
-nxt_port_write(nxt_task_t *task, nxt_cycle_t *cycle, nxt_uint_t type,
+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)
 {
-    nxt_uint_t  i, n;
-    nxt_port_t  *port;
+    nxt_uint_t     i, n, nprocesses, nports;
+    nxt_port_t     *port;
+    nxt_process_t  *process;
 
-    port = cycle->ports->elts;
-    n = cycle->ports->nelts;
+    process = rt->processes->elts;
+    nprocesses = rt->processes->nelts;
 
-    for (i = 0; i < n; i++) {
-        if (nxt_pid != port[i].pid) {
-            (void) nxt_port_socket_write(task, &port[i], type, fd, stream, b);
+    for (i = 0; i < nprocesses; i++) {
+
+        if (nxt_pid != process[i].pid) {
+            port = process[i].ports->elts;
+            nports = process[i].ports->nelts;
+
+            for (n = 0; n < nports; n++) {
+                (void) nxt_port_socket_write(task, &port[n], type,
+                                             fd, stream, b);
+            }
         }
     }
 }
@@ -70,20 +78,21 @@ nxt_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 void
 nxt_port_quit_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 {
-    nxt_cycle_quit(task, NULL);
+    nxt_runtime_quit(task);
 }
 
 
 void
-nxt_port_send_new_port(nxt_task_t *task, nxt_cycle_t *cycle,
+nxt_port_send_new_port(nxt_task_t *task, nxt_runtime_t *rt,
     nxt_port_t *new_port)
 {
     nxt_buf_t                *b;
     nxt_uint_t               i, n;
     nxt_port_t               *port;
+    nxt_process_t            *process;
     nxt_port_msg_new_port_t  *msg;
 
-    n = cycle->ports->nelts;
+    n = rt->processes->nelts;
     if (n == 0) {
         return;
     }
@@ -91,34 +100,33 @@ nxt_port_send_new_port(nxt_task_t *task, nxt_cycle_t *cycle,
     nxt_debug(task, "new port %d for process %PI engine %uD",
               new_port->socket.fd, new_port->pid, new_port->engine);
 
-    port = cycle->ports->elts;
+    process = rt->processes->elts;
 
     for (i = 0; i < n; i++) {
 
-        if (port[i].pid == new_port->pid
-            || port[i].pid == nxt_pid
-            || port[i].engine != 0)
-        {
+        if (process[i].pid == new_port->pid || process[i].pid == nxt_pid) {
             continue;
         }
 
-        b = nxt_buf_mem_alloc(port[i].mem_pool, sizeof(nxt_port_data_t), 0);
+        port = process[i].ports->elts;
+
+        b = nxt_buf_mem_alloc(port->mem_pool, sizeof(nxt_port_data_t), 0);
 
         if (nxt_slow_path(b == NULL)) {
             continue;
         }
 
-        b->data = &port[i];
+        b->data = port;
         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->pid = new_port->pid;
         msg->engine = new_port->engine;
-        msg->max_size = port[i].max_size;
-        msg->max_share = port[i].max_share;
+        msg->max_size = port->max_size;
+        msg->max_share = port->max_share;
 
-        (void) nxt_port_socket_write(task, &port[i], NXT_PORT_MSG_NEW_PORT,
+        (void) nxt_port_socket_write(task, port, NXT_PORT_MSG_NEW_PORT,
                                      new_port->socket.fd, 0, b);
     }
 }
@@ -143,13 +151,19 @@ void
 nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 {
     nxt_port_t               *port;
-    nxt_cycle_t              *cycle;
+    nxt_process_t            *process;
+    nxt_runtime_t            *rt;
     nxt_mem_pool_t           *mp;
     nxt_port_msg_new_port_t  *new_port_msg;
 
-    cycle = nxt_thread_cycle();
+    rt = task->thread->runtime;
 
-    port = nxt_array_add(cycle->ports);
+    process = nxt_runtime_new_process(rt);
+    if (nxt_slow_path(process == NULL)) {
+        return;
+    }
+
+    port = nxt_array_zero_add(process->ports);
     if (nxt_slow_path(port == NULL)) {
         return;
     }
@@ -167,6 +181,8 @@ nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
     nxt_debug(task, "new port %d received for process %PI engine %uD",
               msg->fd, new_port_msg->pid, new_port_msg->engine);
 
+    process->pid = new_port_msg->pid;
+
     port->pid = new_port_msg->pid;
     port->engine = new_port_msg->engine;
     port->pair[0] = -1;
@@ -183,27 +199,29 @@ nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 
 
 void
-nxt_port_change_log_file(nxt_task_t *task, nxt_cycle_t *cycle,
-    nxt_uint_t slot, nxt_fd_t fd)
+nxt_port_change_log_file(nxt_task_t *task, nxt_runtime_t *rt, nxt_uint_t slot,
+    nxt_fd_t fd)
 {
-    nxt_buf_t   *b;
-    nxt_uint_t  i, n;
-    nxt_port_t  *port;
+    nxt_buf_t      *b;
+    nxt_uint_t     i, n;
+    nxt_port_t     *port;
+    nxt_process_t  *process;
 
-    n = cycle->ports->nelts;
+    n = rt->processes->nelts;
     if (n == 0) {
         return;
     }
 
     nxt_debug(task, "change log file #%ui fd:%FD", slot, fd);
 
-    port = cycle->ports->elts;
+    process = rt->processes->elts;
 
-    /* port[0] is master process port. */
+    /* process[0] is master process. */
 
     for (i = 1; i < n; i++) {
-        b = nxt_buf_mem_alloc(port[i].mem_pool, sizeof(nxt_port_data_t), 0);
+        port = process[i].ports->elts;
 
+        b = nxt_buf_mem_alloc(port->mem_pool, sizeof(nxt_port_data_t), 0);
         if (nxt_slow_path(b == NULL)) {
             continue;
         }
@@ -211,7 +229,7 @@ nxt_port_change_log_file(nxt_task_t *task, nxt_cycle_t *cycle,
         *(nxt_uint_t *) b->mem.pos = slot;
         b->mem.free += sizeof(nxt_uint_t);
 
-        (void) nxt_port_socket_write(task, &port[i], NXT_PORT_MSG_CHANGE_FILE,
+        (void) nxt_port_socket_write(task, port, NXT_PORT_MSG_CHANGE_FILE,
                                      fd, 0, b);
     }
 }
@@ -220,17 +238,17 @@ nxt_port_change_log_file(nxt_task_t *task, nxt_cycle_t *cycle,
 void
 nxt_port_change_log_file_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 {
-    nxt_buf_t    *b;
-    nxt_uint_t   slot;
-    nxt_file_t   *log_file;
-    nxt_cycle_t  *cycle;
+    nxt_buf_t      *b;
+    nxt_uint_t     slot;
+    nxt_file_t     *log_file;
+    nxt_runtime_t  *rt;
 
-    cycle = nxt_thread_cycle();
+    rt = task->thread->runtime;
 
     b = msg->buf;
     slot = *(nxt_uint_t *) b->mem.pos;
 
-    log_file = nxt_list_elt(cycle->log_files, slot);
+    log_file = nxt_list_elt(rt->log_files, slot);
 
     nxt_debug(task, "change log file %FD:%FD", msg->fd, log_file->fd);