1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
|
/*
* Copyright (C) Igor Sysoev
* Copyright (C) NGINX, Inc.
*/
#ifndef _NXT_PORT_H_INCLUDED_
#define _NXT_PORT_H_INCLUDED_
struct nxt_port_handlers_s {
/* RPC responses. */
nxt_port_handler_t rpc_ready;
nxt_port_handler_t rpc_error;
/* Main process RPC requests. */
nxt_port_handler_t start_worker;
nxt_port_handler_t socket;
nxt_port_handler_t modules;
nxt_port_handler_t conf_store;
/* File descriptor exchange. */
nxt_port_handler_t change_file;
nxt_port_handler_t new_port;
nxt_port_handler_t mmap;
/* New process ready. */
nxt_port_handler_t process_ready;
/* Process exit/crash notification. */
nxt_port_handler_t remove_pid;
/* Stop process command. */
nxt_port_handler_t quit;
/* Various data. */
nxt_port_handler_t data;
};
#define nxt_port_handler_idx(name) \
( offsetof(nxt_port_handlers_t, name) / sizeof(nxt_port_handler_t) )
typedef enum {
NXT_PORT_MSG_LAST = 0x100,
NXT_PORT_MSG_CLOSE_FD = 0x200,
NXT_PORT_MSG_SYNC = 0x400,
NXT_PORT_MSG_MASK = 0xFF,
_NXT_PORT_MSG_RPC_READY = nxt_port_handler_idx(rpc_ready),
_NXT_PORT_MSG_RPC_ERROR = nxt_port_handler_idx(rpc_error),
_NXT_PORT_MSG_START_WORKER = nxt_port_handler_idx(start_worker),
_NXT_PORT_MSG_SOCKET = nxt_port_handler_idx(socket),
_NXT_PORT_MSG_MODULES = nxt_port_handler_idx(modules),
_NXT_PORT_MSG_CONF_STORE = nxt_port_handler_idx(conf_store),
_NXT_PORT_MSG_CHANGE_FILE = nxt_port_handler_idx(change_file),
_NXT_PORT_MSG_NEW_PORT = nxt_port_handler_idx(new_port),
_NXT_PORT_MSG_MMAP = nxt_port_handler_idx(mmap),
_NXT_PORT_MSG_PROCESS_READY = nxt_port_handler_idx(process_ready),
_NXT_PORT_MSG_REMOVE_PID = nxt_port_handler_idx(remove_pid),
_NXT_PORT_MSG_QUIT = nxt_port_handler_idx(quit),
_NXT_PORT_MSG_DATA = nxt_port_handler_idx(data),
NXT_PORT_MSG_MAX = sizeof(nxt_port_handlers_t) /
sizeof(nxt_port_handler_t),
NXT_PORT_MSG_RPC_READY = _NXT_PORT_MSG_RPC_READY,
NXT_PORT_MSG_RPC_READY_LAST = _NXT_PORT_MSG_RPC_READY | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_RPC_ERROR = _NXT_PORT_MSG_RPC_ERROR | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_START_WORKER = _NXT_PORT_MSG_START_WORKER |
NXT_PORT_MSG_LAST,
NXT_PORT_MSG_SOCKET = _NXT_PORT_MSG_SOCKET | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_MODULES = _NXT_PORT_MSG_MODULES | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_CONF_STORE = _NXT_PORT_MSG_CONF_STORE | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_CHANGE_FILE = _NXT_PORT_MSG_CHANGE_FILE | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_NEW_PORT = _NXT_PORT_MSG_NEW_PORT | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_MMAP = _NXT_PORT_MSG_MMAP | NXT_PORT_MSG_LAST |
NXT_PORT_MSG_CLOSE_FD | NXT_PORT_MSG_SYNC,
NXT_PORT_MSG_PROCESS_READY = _NXT_PORT_MSG_PROCESS_READY |
NXT_PORT_MSG_LAST,
NXT_PORT_MSG_QUIT = _NXT_PORT_MSG_QUIT | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_REMOVE_PID = _NXT_PORT_MSG_REMOVE_PID | NXT_PORT_MSG_LAST,
NXT_PORT_MSG_DATA = _NXT_PORT_MSG_DATA,
NXT_PORT_MSG_DATA_LAST = _NXT_PORT_MSG_DATA | NXT_PORT_MSG_LAST,
} nxt_port_msg_type_t;
/* Passed as a first iov chunk. */
typedef struct {
uint32_t stream;
nxt_pid_t pid;
nxt_port_id_t reply_port;
uint8_t type;
/* Last message for this stream. */
uint8_t last; /* 1 bit */
/* Message data send using mmap, next chunk is a nxt_port_mmap_msg_t. */
uint8_t mmap; /* 1 bit */
/* Non-First fragment in fragmented message sequence. */
uint8_t nf; /* 1 bit */
/* More Fragments followed. */
uint8_t mf; /* 1 bit */
/* Message delivery tracking enabled, next chunk is tracking msg. */
uint8_t tracking; /* 1 bit */
} nxt_port_msg_t;
typedef struct {
nxt_queue_link_t link;
nxt_buf_t *buf;
size_t share;
nxt_fd_t fd;
nxt_bool_t close_fd;
nxt_port_msg_t port_msg;
uint32_t tracking_msg[2];
nxt_work_t work;
} nxt_port_send_msg_t;
struct nxt_port_recv_msg_s {
nxt_fd_t fd;
nxt_buf_t *buf;
nxt_port_t *port;
nxt_port_msg_t port_msg;
size_t size;
nxt_bool_t cancelled;
union {
nxt_port_t *new_port;
nxt_pid_t removed_pid;
void *data;
} u;
};
typedef struct nxt_app_s nxt_app_t;
struct nxt_port_s {
nxt_fd_event_t socket;
nxt_queue_link_t link; /* for nxt_process_t.ports */
nxt_process_t *process;
nxt_queue_link_t app_link; /* for nxt_app_t.ports */
nxt_app_t *app;
nxt_queue_t messages; /* of nxt_port_send_msg_t */
nxt_thread_mutex_t write_mutex;
/* Maximum size of message part. */
uint32_t max_size;
/* Maximum interleave of message parts. */
uint32_t max_share;
uint32_t app_pending_responses;
uint32_t app_responses;
nxt_port_handler_t handler;
nxt_port_handler_t *data;
nxt_mp_t *mem_pool;
nxt_event_engine_t *engine;
nxt_buf_t *free_bufs;
nxt_socket_t pair[2];
nxt_port_id_t id;
nxt_pid_t pid;
nxt_lvlhsh_t rpc_streams; /* stream to nxt_port_rpc_reg_t */
nxt_lvlhsh_t rpc_peers; /* peer to queue of nxt_port_rpc_reg_t */
nxt_lvlhsh_t frags;
nxt_atomic_t use_count;
nxt_process_type_t type;
struct iovec *iov;
void *mmsg_buf;
};
typedef struct {
nxt_port_id_t id;
nxt_pid_t pid;
size_t max_size;
size_t max_share;
nxt_process_type_t type:8;
} nxt_port_msg_new_port_t;
/*
* nxt_port_data_t size is allocation size
* which enables effective reuse of memory pool cache.
*/
typedef union {
nxt_buf_t buf;
nxt_port_msg_new_port_t new_port;
} nxt_port_data_t;
typedef void (*nxt_port_post_handler_t)(nxt_task_t *task, nxt_port_t *port,
void *data);
nxt_port_t *nxt_port_new(nxt_task_t *task, nxt_port_id_t id, nxt_pid_t pid,
nxt_process_type_t type);
nxt_port_id_t nxt_port_get_next_id(void);
void nxt_port_reset_next_id(void);
nxt_int_t nxt_port_socket_init(nxt_task_t *task, nxt_port_t *port,
size_t max_size);
void nxt_port_destroy(nxt_port_t *port);
void nxt_port_close(nxt_task_t *task, nxt_port_t *port);
void nxt_port_write_enable(nxt_task_t *task, nxt_port_t *port);
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_twrite(nxt_task_t *task, nxt_port_t *port,
nxt_uint_t type, nxt_fd_t fd, uint32_t stream, nxt_port_id_t reply_port,
nxt_buf_t *b, void *tracking);
nxt_inline 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_port_id_t reply_port,
nxt_buf_t *b)
{
return nxt_port_socket_twrite(task, port, type, fd, stream, reply_port, b,
NULL);
}
void nxt_port_enable(nxt_task_t *task, nxt_port_t *port,
nxt_port_handlers_t *handlers);
nxt_int_t nxt_port_send_port(nxt_task_t *task, nxt_port_t *port,
nxt_port_t *new_port, uint32_t stream);
void nxt_port_change_log_file(nxt_task_t *task, nxt_runtime_t *rt,
nxt_uint_t slot, nxt_fd_t fd);
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_process_ready_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_remove_pid_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);
nxt_int_t nxt_port_post(nxt_task_t *task, nxt_port_t *port,
nxt_port_post_handler_t handler, void *data);
void nxt_port_use(nxt_task_t *task, nxt_port_t *port, int i);
#endif /* _NXT_PORT_H_INCLUDED_ */
|
