Using shared memory to send data via nxt_port.

Usage:
    b = nxt_port_mmap_get_buf(task, port, size);
    b->mem.free = nxt_cpymem(b->mem.free, data, size);
    nxt_port_socket_write(task, port, NXT_PORT_MSG_DATA, -1, 0, b);

1 files changed, 32 insertions, 17 deletions

diff --git a/src/nxt_port.h b/src/nxt_port.h
index 8a9b8926..76775660 100644
--- a/src/nxt_port.h
+++ b/src/nxt_port.h
@@ -8,11 +8,28 @@
 #define _NXT_PORT_H_INCLUDED_
 
 
+typedef enum {
+    NXT_PORT_MSG_QUIT = 0,
+    NXT_PORT_MSG_NEW_PORT,
+    NXT_PORT_MSG_CHANGE_FILE,
+    NXT_PORT_MSG_MMAP,
+    NXT_PORT_MSG_DATA,
+} nxt_port_msg_type_t;
+
+#define NXT_PORT_MSG_MAX  NXT_PORT_MSG_DATA
+
+
+/* Passed as a first iov chunk. */
 typedef struct {
-    uint32_t            stream;
+    uint32_t             stream;
+    nxt_pid_t            pid;
+    nxt_port_id_t        reply_port;
 
-    uint16_t            type;
-    uint8_t             last;      /* 1 bit */
+    nxt_port_msg_type_t  type:8;
+    uint8_t              last;      /* 1 bit */
+
+    /* Message data send using mmap, next chunk is a nxt_port_mmap_msg_t. */
+    uint8_t              mmap;      /* 1 bit */
 } nxt_port_msg_t;
 
 
@@ -26,12 +43,10 @@ typedef struct {
 
 
 struct nxt_port_recv_msg_s {
-    uint32_t            stream;
-    uint16_t            type;
-
     nxt_fd_t            fd;
     nxt_buf_t           *buf;
     nxt_port_t          *port;
+    nxt_port_msg_t      port_msg;
 };
 
 
@@ -39,6 +54,8 @@ struct nxt_port_s {
     /* Must be the first field. */
     nxt_fd_event_t      socket;
 
+    nxt_queue_link_t    link;
+
     nxt_queue_t         messages;   /* of nxt_port_send_msg_t */
 
     /* Maximum size of message part. */
@@ -53,26 +70,22 @@ struct nxt_port_s {
     nxt_buf_t           *free_bufs;
     nxt_socket_t        pair[2];
 
+    nxt_port_id_t       id;
     nxt_pid_t           pid;
     uint32_t            engine;
-};
-
-
-#define NXT_PORT_MSG_MAX  NXT_PORT_MSG_DATA
 
-typedef enum {
-    NXT_PORT_MSG_QUIT = 0,
-    NXT_PORT_MSG_NEW_PORT,
-    NXT_PORT_MSG_CHANGE_FILE,
-    NXT_PORT_MSG_DATA,
-} nxt_port_msg_type_e;
+    nxt_process_type_t  type:8;
+    nxt_process_t       *process;
+};
 
 
 typedef struct {
+    nxt_port_id_t       id;
     nxt_pid_t           pid;
     uint32_t            engine;
     size_t              max_size;
     size_t              max_share;
+    nxt_process_type_t  type:8;
 } nxt_port_msg_new_port_t;
 
 
@@ -94,7 +107,8 @@ void nxt_port_write_close(nxt_port_t *port);
 void nxt_port_read_enable(nxt_task_t *task, nxt_port_t *port);
 void nxt_port_read_close(nxt_port_t *port);
 nxt_int_t nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port,
-    nxt_uint_t type, nxt_fd_t fd, uint32_t stream, nxt_buf_t *b);
+    nxt_uint_t type, nxt_fd_t fd, uint32_t stream, nxt_port_id_t reply_port,
+    nxt_buf_t *b);
 
 void nxt_port_create(nxt_thread_t *thread, nxt_port_t *port,
     nxt_port_handler_t *handlers);
@@ -109,6 +123,7 @@ void nxt_port_quit_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
 void nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
 void nxt_port_change_log_file_handler(nxt_task_t *task,
     nxt_port_recv_msg_t *msg);
+void nxt_port_mmap_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
 void nxt_port_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
 void nxt_port_empty_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);