xref: /unit/src/nxt_h1proto.c (revision 1418:073c9c008b83)
1 
2 /*
3  * Copyright (C) Igor Sysoev
4  * Copyright (C) NGINX, Inc.
5  */
6 
7 #include <nxt_router.h>
8 #include <nxt_http.h>
9 #include <nxt_upstream.h>
10 #include <nxt_h1proto.h>
11 #include <nxt_websocket.h>
12 #include <nxt_websocket_header.h>
13 
14 
15 /*
16  * nxt_http_conn_ and nxt_h1p_conn_ prefixes are used for connection handlers.
17  * nxt_h1p_idle_ prefix is used for idle connection handlers.
18  * nxt_h1p_request_ prefix is used for HTTP/1 protocol request methods.
19  */
20 
21 #if (NXT_TLS)
22 static ssize_t nxt_http_idle_io_read_handler(nxt_task_t *task, nxt_conn_t *c);
23 static void nxt_http_conn_test(nxt_task_t *task, void *obj, void *data);
24 #endif
25 static ssize_t nxt_h1p_idle_io_read_handler(nxt_task_t *task, nxt_conn_t *c);
26 static void nxt_h1p_conn_proto_init(nxt_task_t *task, void *obj, void *data);
27 static void nxt_h1p_conn_request_init(nxt_task_t *task, void *obj, void *data);
28 static void nxt_h1p_conn_request_header_parse(nxt_task_t *task, void *obj,
29     void *data);
30 static nxt_int_t nxt_h1p_header_process(nxt_task_t *task, nxt_h1proto_t *h1p,
31     nxt_http_request_t *r);
32 static nxt_int_t nxt_h1p_header_buffer_test(nxt_task_t *task,
33     nxt_h1proto_t *h1p, nxt_conn_t *c, nxt_socket_conf_t *skcf);
34 static nxt_int_t nxt_h1p_connection(void *ctx, nxt_http_field_t *field,
35     uintptr_t data);
36 static nxt_int_t nxt_h1p_upgrade(void *ctx, nxt_http_field_t *field,
37     uintptr_t data);
38 static nxt_int_t nxt_h1p_websocket_key(void *ctx, nxt_http_field_t *field,
39     uintptr_t data);
40 static nxt_int_t nxt_h1p_websocket_version(void *ctx, nxt_http_field_t *field,
41     uintptr_t data);
42 static nxt_int_t nxt_h1p_transfer_encoding(void *ctx, nxt_http_field_t *field,
43     uintptr_t data);
44 static void nxt_h1p_request_body_read(nxt_task_t *task, nxt_http_request_t *r);
45 static void nxt_h1p_conn_request_body_read(nxt_task_t *task, void *obj,
46     void *data);
47 static void nxt_h1p_request_local_addr(nxt_task_t *task, nxt_http_request_t *r);
48 static void nxt_h1p_request_header_send(nxt_task_t *task,
49     nxt_http_request_t *r, nxt_work_handler_t body_handler, void *data);
50 static void nxt_h1p_request_send(nxt_task_t *task, nxt_http_request_t *r,
51     nxt_buf_t *out);
52 static nxt_buf_t *nxt_h1p_chunk_create(nxt_task_t *task, nxt_http_request_t *r,
53     nxt_buf_t *out);
54 static nxt_off_t nxt_h1p_request_body_bytes_sent(nxt_task_t *task,
55     nxt_http_proto_t proto);
56 static void nxt_h1p_request_discard(nxt_task_t *task, nxt_http_request_t *r,
57     nxt_buf_t *last);
58 static void nxt_h1p_conn_request_error(nxt_task_t *task, void *obj, void *data);
59 static void nxt_h1p_conn_request_timeout(nxt_task_t *task, void *obj,
60     void *data);
61 static void nxt_h1p_conn_request_send_timeout(nxt_task_t *task, void *obj,
62     void *data);
63 nxt_inline void nxt_h1p_request_error(nxt_task_t *task, nxt_h1proto_t *h1p,
64     nxt_http_request_t *r);
65 static void nxt_h1p_request_close(nxt_task_t *task, nxt_http_proto_t proto,
66     nxt_socket_conf_joint_t *joint);
67 static void nxt_h1p_conn_sent(nxt_task_t *task, void *obj, void *data);
68 static void nxt_h1p_conn_close(nxt_task_t *task, void *obj, void *data);
69 static void nxt_h1p_conn_error(nxt_task_t *task, void *obj, void *data);
70 static nxt_msec_t nxt_h1p_conn_timer_value(nxt_conn_t *c, uintptr_t data);
71 static void nxt_h1p_keepalive(nxt_task_t *task, nxt_h1proto_t *h1p,
72     nxt_conn_t *c);
73 static void nxt_h1p_idle_close(nxt_task_t *task, void *obj, void *data);
74 static void nxt_h1p_idle_timeout(nxt_task_t *task, void *obj, void *data);
75 static void nxt_h1p_idle_response(nxt_task_t *task, nxt_conn_t *c);
76 static void nxt_h1p_idle_response_sent(nxt_task_t *task, void *obj, void *data);
77 static void nxt_h1p_idle_response_timeout(nxt_task_t *task, void *obj,
78     void *data);
79 static nxt_msec_t nxt_h1p_idle_response_timer_value(nxt_conn_t *c,
80     uintptr_t data);
81 static void nxt_h1p_shutdown(nxt_task_t *task, nxt_conn_t *c);
82 static void nxt_h1p_closing(nxt_task_t *task, nxt_conn_t *c);
83 static void nxt_h1p_conn_ws_shutdown(nxt_task_t *task, void *obj, void *data);
84 static void nxt_h1p_conn_closing(nxt_task_t *task, void *obj, void *data);
85 static void nxt_h1p_conn_free(nxt_task_t *task, void *obj, void *data);
86 
87 static void nxt_h1p_peer_connect(nxt_task_t *task, nxt_http_peer_t *peer);
88 static void nxt_h1p_peer_connected(nxt_task_t *task, void *obj, void *data);
89 static void nxt_h1p_peer_refused(nxt_task_t *task, void *obj, void *data);
90 static void nxt_h1p_peer_header_send(nxt_task_t *task, nxt_http_peer_t *peer);
91 static void nxt_h1p_peer_header_sent(nxt_task_t *task, void *obj, void *data);
92 static void nxt_h1p_peer_header_read(nxt_task_t *task, nxt_http_peer_t *peer);
93 static ssize_t nxt_h1p_peer_io_read_handler(nxt_task_t *task, nxt_conn_t *c);
94 static void nxt_h1p_peer_header_read_done(nxt_task_t *task, void *obj,
95     void *data);
96 static nxt_int_t nxt_h1p_peer_header_parse(nxt_http_peer_t *peer,
97     nxt_buf_mem_t *bm);
98 static void nxt_h1p_peer_read(nxt_task_t *task, nxt_http_peer_t *peer);
99 static void nxt_h1p_peer_read_done(nxt_task_t *task, void *obj, void *data);
100 static void nxt_h1p_peer_closed(nxt_task_t *task, void *obj, void *data);
101 static void nxt_h1p_peer_error(nxt_task_t *task, void *obj, void *data);
102 static void nxt_h1p_peer_send_timeout(nxt_task_t *task, void *obj, void *data);
103 static void nxt_h1p_peer_read_timeout(nxt_task_t *task, void *obj, void *data);
104 static nxt_msec_t nxt_h1p_peer_timer_value(nxt_conn_t *c, uintptr_t data);
105 static void nxt_h1p_peer_close(nxt_task_t *task, nxt_http_peer_t *peer);
106 static void nxt_h1p_peer_free(nxt_task_t *task, void *obj, void *data);
107 
108 #if (NXT_TLS)
109 static const nxt_conn_state_t  nxt_http_idle_state;
110 static const nxt_conn_state_t  nxt_h1p_shutdown_state;
111 #endif
112 static const nxt_conn_state_t  nxt_h1p_idle_state;
113 static const nxt_conn_state_t  nxt_h1p_header_parse_state;
114 static const nxt_conn_state_t  nxt_h1p_read_body_state;
115 static const nxt_conn_state_t  nxt_h1p_request_send_state;
116 static const nxt_conn_state_t  nxt_h1p_timeout_response_state;
117 static const nxt_conn_state_t  nxt_h1p_keepalive_state;
118 static const nxt_conn_state_t  nxt_h1p_close_state;
119 static const nxt_conn_state_t  nxt_h1p_peer_connect_state;
120 static const nxt_conn_state_t  nxt_h1p_peer_header_send_state;
121 static const nxt_conn_state_t  nxt_h1p_peer_header_body_send_state;
122 static const nxt_conn_state_t  nxt_h1p_peer_header_read_state;
123 static const nxt_conn_state_t  nxt_h1p_peer_header_read_timer_state;
124 static const nxt_conn_state_t  nxt_h1p_peer_read_state;
125 static const nxt_conn_state_t  nxt_h1p_peer_close_state;
126 
127 
128 const nxt_http_proto_table_t  nxt_http_proto[3] = {
129     /* NXT_HTTP_PROTO_H1 */
130     {
131         .body_read        = nxt_h1p_request_body_read,
132         .local_addr       = nxt_h1p_request_local_addr,
133         .header_send      = nxt_h1p_request_header_send,
134         .send             = nxt_h1p_request_send,
135         .body_bytes_sent  = nxt_h1p_request_body_bytes_sent,
136         .discard          = nxt_h1p_request_discard,
137         .close            = nxt_h1p_request_close,
138 
139         .peer_connect     = nxt_h1p_peer_connect,
140         .peer_header_send = nxt_h1p_peer_header_send,
141         .peer_header_read = nxt_h1p_peer_header_read,
142         .peer_read        = nxt_h1p_peer_read,
143         .peer_close       = nxt_h1p_peer_close,
144 
145         .ws_frame_start   = nxt_h1p_websocket_frame_start,
146     },
147     /* NXT_HTTP_PROTO_H2      */
148     /* NXT_HTTP_PROTO_DEVNULL */
149 };
150 
151 
152 static nxt_lvlhsh_t                    nxt_h1p_fields_hash;
153 
154 static nxt_http_field_proc_t           nxt_h1p_fields[] = {
155     { nxt_string("Connection"),        &nxt_h1p_connection, 0 },
156     { nxt_string("Upgrade"),           &nxt_h1p_upgrade, 0 },
157     { nxt_string("Sec-WebSocket-Key"), &nxt_h1p_websocket_key, 0 },
158     { nxt_string("Sec-WebSocket-Version"),
159                                        &nxt_h1p_websocket_version, 0 },
160     { nxt_string("Transfer-Encoding"), &nxt_h1p_transfer_encoding, 0 },
161 
162     { nxt_string("Host"),              &nxt_http_request_host, 0 },
163     { nxt_string("Cookie"),            &nxt_http_request_field,
164         offsetof(nxt_http_request_t, cookie) },
165     { nxt_string("Referer"),           &nxt_http_request_field,
166         offsetof(nxt_http_request_t, referer) },
167     { nxt_string("User-Agent"),        &nxt_http_request_field,
168         offsetof(nxt_http_request_t, user_agent) },
169     { nxt_string("Content-Type"),      &nxt_http_request_field,
170         offsetof(nxt_http_request_t, content_type) },
171     { nxt_string("Content-Length"),    &nxt_http_request_content_length, 0 },
172 };
173 
174 
175 static nxt_lvlhsh_t                    nxt_h1p_peer_fields_hash;
176 
177 static nxt_http_field_proc_t           nxt_h1p_peer_fields[] = {
178     { nxt_string("Connection"),        &nxt_http_proxy_skip, 0 },
179     { nxt_string("Transfer-Encoding"), &nxt_http_proxy_skip, 0 },
180     { nxt_string("Server"),            &nxt_http_proxy_skip, 0 },
181     { nxt_string("Date"),              &nxt_http_proxy_date, 0 },
182     { nxt_string("Content-Length"),    &nxt_http_proxy_content_length, 0 },
183 };
184 
185 
186 nxt_int_t
187 nxt_h1p_init(nxt_task_t *task, nxt_runtime_t *rt)
188 {
189     nxt_int_t  ret;
190 
191     ret = nxt_http_fields_hash(&nxt_h1p_fields_hash, rt->mem_pool,
192                                nxt_h1p_fields, nxt_nitems(nxt_h1p_fields));
193 
194     if (nxt_fast_path(ret == NXT_OK)) {
195         ret = nxt_http_fields_hash(&nxt_h1p_peer_fields_hash,
196                                    rt->mem_pool, nxt_h1p_peer_fields,
197                                    nxt_nitems(nxt_h1p_peer_fields));
198     }
199 
200     return ret;
201 }
202 
203 
204 void
205 nxt_http_conn_init(nxt_task_t *task, void *obj, void *data)
206 {
207     nxt_conn_t               *c;
208     nxt_socket_conf_t        *skcf;
209     nxt_event_engine_t       *engine;
210     nxt_listen_event_t       *lev;
211     nxt_socket_conf_joint_t  *joint;
212 
213     c = obj;
214     lev = data;
215 
216     nxt_debug(task, "http conn init");
217 
218     joint = lev->socket.data;
219     skcf = joint->socket_conf;
220     c->local = skcf->sockaddr;
221 
222     engine = task->thread->engine;
223     c->read_work_queue = &engine->fast_work_queue;
224     c->write_work_queue = &engine->fast_work_queue;
225 
226     c->read_state = &nxt_h1p_idle_state;
227 
228 #if (NXT_TLS)
229     if (skcf->tls != NULL) {
230         c->read_state = &nxt_http_idle_state;
231     }
232 #endif
233 
234     nxt_conn_read(engine, c);
235 }
236 
237 
238 #if (NXT_TLS)
239 
240 static const nxt_conn_state_t  nxt_http_idle_state
241     nxt_aligned(64) =
242 {
243     .ready_handler = nxt_http_conn_test,
244     .close_handler = nxt_h1p_conn_close,
245     .error_handler = nxt_h1p_conn_error,
246 
247     .io_read_handler = nxt_http_idle_io_read_handler,
248 
249     .timer_handler = nxt_h1p_idle_timeout,
250     .timer_value = nxt_h1p_conn_timer_value,
251     .timer_data = offsetof(nxt_socket_conf_t, idle_timeout),
252 };
253 
254 
255 static ssize_t
256 nxt_http_idle_io_read_handler(nxt_task_t *task, nxt_conn_t *c)
257 {
258     size_t                   size;
259     ssize_t                  n;
260     nxt_buf_t                *b;
261     nxt_socket_conf_joint_t  *joint;
262 
263     joint = c->listen->socket.data;
264 
265     if (nxt_slow_path(joint == NULL)) {
266         /*
267          * Listening socket had been closed while
268          * connection was in keep-alive state.
269          */
270         c->read_state = &nxt_h1p_idle_close_state;
271         return 0;
272     }
273 
274     size = joint->socket_conf->header_buffer_size;
275 
276     b = nxt_event_engine_buf_mem_alloc(task->thread->engine, size);
277     if (nxt_slow_path(b == NULL)) {
278         c->socket.error = NXT_ENOMEM;
279         return NXT_ERROR;
280     }
281 
282     /*
283      * 1 byte is enough to distinguish between SSLv3/TLS and plain HTTP.
284      * 11 bytes are enough to log supported SSLv3/TLS version.
285      * 16 bytes are just for more optimized kernel copy-out operation.
286      */
287     n = c->io->recv(c, b->mem.pos, 16, MSG_PEEK);
288 
289     if (n > 0) {
290         c->read = b;
291 
292     } else {
293         c->read = NULL;
294         nxt_event_engine_buf_mem_free(task->thread->engine, b);
295     }
296 
297     return n;
298 }
299 
300 
301 static void
302 nxt_http_conn_test(nxt_task_t *task, void *obj, void *data)
303 {
304     u_char                   *p;
305     nxt_buf_t                *b;
306     nxt_conn_t               *c;
307     nxt_tls_conf_t           *tls;
308     nxt_event_engine_t       *engine;
309     nxt_socket_conf_joint_t  *joint;
310 
311     c = obj;
312 
313     nxt_debug(task, "h1p conn https test");
314 
315     engine = task->thread->engine;
316     b = c->read;
317     p = b->mem.pos;
318 
319     c->read_state = &nxt_h1p_idle_state;
320 
321     if (p[0] != 0x16) {
322         b->mem.free = b->mem.pos;
323 
324         nxt_conn_read(engine, c);
325         return;
326     }
327 
328     /* SSLv3/TLS ClientHello message. */
329 
330 #if (NXT_DEBUG)
331     if (nxt_buf_mem_used_size(&b->mem) >= 11) {
332         u_char      major, minor;
333         const char  *protocol;
334 
335         major = p[9];
336         minor = p[10];
337 
338         if (major == 3) {
339             if (minor == 0) {
340                 protocol = "SSLv";
341 
342             } else {
343                 protocol = "TLSv";
344                 major -= 2;
345                 minor -= 1;
346             }
347 
348             nxt_debug(task, "SSL/TLS: %s%ud.%ud", protocol, major, minor);
349         }
350     }
351 #endif
352 
353     c->read = NULL;
354     nxt_event_engine_buf_mem_free(engine, b);
355 
356     joint = c->listen->socket.data;
357 
358     if (nxt_slow_path(joint == NULL)) {
359         /*
360          * Listening socket had been closed while
361          * connection was in keep-alive state.
362          */
363         nxt_h1p_closing(task, c);
364         return;
365     }
366 
367     tls = joint->socket_conf->tls;
368 
369     tls->conn_init(task, tls, c);
370 }
371 
372 #endif
373 
374 
375 static const nxt_conn_state_t  nxt_h1p_idle_state
376     nxt_aligned(64) =
377 {
378     .ready_handler = nxt_h1p_conn_proto_init,
379     .close_handler = nxt_h1p_conn_close,
380     .error_handler = nxt_h1p_conn_error,
381 
382     .io_read_handler = nxt_h1p_idle_io_read_handler,
383 
384     .timer_handler = nxt_h1p_idle_timeout,
385     .timer_value = nxt_h1p_conn_timer_value,
386     .timer_data = offsetof(nxt_socket_conf_t, idle_timeout),
387     .timer_autoreset = 1,
388 };
389 
390 
391 static ssize_t
392 nxt_h1p_idle_io_read_handler(nxt_task_t *task, nxt_conn_t *c)
393 {
394     size_t                   size;
395     ssize_t                  n;
396     nxt_buf_t                *b;
397     nxt_socket_conf_joint_t  *joint;
398 
399     joint = c->listen->socket.data;
400 
401     if (nxt_slow_path(joint == NULL)) {
402         /*
403          * Listening socket had been closed while
404          * connection was in keep-alive state.
405          */
406         c->read_state = &nxt_h1p_idle_close_state;
407         return 0;
408     }
409 
410     b = c->read;
411 
412     if (b == NULL) {
413         size = joint->socket_conf->header_buffer_size;
414 
415         b = nxt_event_engine_buf_mem_alloc(task->thread->engine, size);
416         if (nxt_slow_path(b == NULL)) {
417             c->socket.error = NXT_ENOMEM;
418             return NXT_ERROR;
419         }
420     }
421 
422     n = c->io->recvbuf(c, b);
423 
424     if (n > 0) {
425         c->read = b;
426 
427     } else {
428         c->read = NULL;
429         nxt_event_engine_buf_mem_free(task->thread->engine, b);
430     }
431 
432     return n;
433 }
434 
435 
436 static void
437 nxt_h1p_conn_proto_init(nxt_task_t *task, void *obj, void *data)
438 {
439     nxt_conn_t     *c;
440     nxt_h1proto_t  *h1p;
441 
442     c = obj;
443 
444     nxt_debug(task, "h1p conn proto init");
445 
446     h1p = nxt_mp_zget(c->mem_pool, sizeof(nxt_h1proto_t));
447     if (nxt_slow_path(h1p == NULL)) {
448         nxt_h1p_closing(task, c);
449         return;
450     }
451 
452     c->socket.data = h1p;
453     h1p->conn = c;
454 
455     nxt_h1p_conn_request_init(task, c, h1p);
456 }
457 
458 
459 static void
460 nxt_h1p_conn_request_init(nxt_task_t *task, void *obj, void *data)
461 {
462     nxt_int_t                ret;
463     nxt_conn_t               *c;
464     nxt_h1proto_t            *h1p;
465     nxt_http_request_t       *r;
466     nxt_socket_conf_joint_t  *joint;
467 
468     c = obj;
469     h1p = data;
470 
471     nxt_debug(task, "h1p conn request init");
472 
473     r = nxt_http_request_create(task);
474 
475     if (nxt_fast_path(r != NULL)) {
476         h1p->request = r;
477         r->proto.h1 = h1p;
478 
479         /* r->protocol = NXT_HTTP_PROTO_H1 is done by zeroing. */
480         r->remote = c->remote;
481 
482 #if (NXT_TLS)
483         r->tls = c->u.tls;
484 #endif
485 
486         r->task = c->task;
487         task = &r->task;
488         c->socket.task = task;
489         c->read_timer.task = task;
490         c->write_timer.task = task;
491 
492         ret = nxt_http_parse_request_init(&h1p->parser, r->mem_pool);
493 
494         if (nxt_fast_path(ret == NXT_OK)) {
495             joint = c->listen->socket.data;
496             joint->count++;
497 
498             r->conf = joint;
499             c->local = joint->socket_conf->sockaddr;
500 
501             nxt_h1p_conn_request_header_parse(task, c, h1p);
502             return;
503         }
504 
505         /*
506          * The request is very incomplete here,
507          * so "internal server error" useless here.
508          */
509         nxt_mp_release(r->mem_pool);
510     }
511 
512     nxt_h1p_closing(task, c);
513 }
514 
515 
516 static const nxt_conn_state_t  nxt_h1p_header_parse_state
517     nxt_aligned(64) =
518 {
519     .ready_handler = nxt_h1p_conn_request_header_parse,
520     .close_handler = nxt_h1p_conn_request_error,
521     .error_handler = nxt_h1p_conn_request_error,
522 
523     .timer_handler = nxt_h1p_conn_request_timeout,
524     .timer_value = nxt_h1p_conn_request_timer_value,
525     .timer_data = offsetof(nxt_socket_conf_t, header_read_timeout),
526 };
527 
528 
529 static void
530 nxt_h1p_conn_request_header_parse(nxt_task_t *task, void *obj, void *data)
531 {
532     nxt_int_t           ret;
533     nxt_conn_t          *c;
534     nxt_h1proto_t       *h1p;
535     nxt_http_status_t   status;
536     nxt_http_request_t  *r;
537 
538     c = obj;
539     h1p = data;
540 
541     nxt_debug(task, "h1p conn header parse");
542 
543     ret = nxt_http_parse_request(&h1p->parser, &c->read->mem);
544 
545     ret = nxt_expect(NXT_DONE, ret);
546 
547     if (ret != NXT_AGAIN) {
548         nxt_timer_disable(task->thread->engine, &c->read_timer);
549     }
550 
551     r = h1p->request;
552 
553     switch (ret) {
554 
555     case NXT_DONE:
556         /*
557          * By default the keepalive mode is disabled in HTTP/1.0 and
558          * enabled in HTTP/1.1.  The mode can be overridden later by
559          * the "Connection" field processed in nxt_h1p_connection().
560          */
561         h1p->keepalive = (h1p->parser.version.s.minor != '0');
562 
563         ret = nxt_h1p_header_process(task, h1p, r);
564 
565         if (nxt_fast_path(ret == NXT_OK)) {
566 
567 #if (NXT_TLS)
568             if (c->u.tls == NULL && r->conf->socket_conf->tls != NULL) {
569                 status = NXT_HTTP_TO_HTTPS;
570                 goto error;
571             }
572 #endif
573 
574             r->state->ready_handler(task, r, NULL);
575             return;
576         }
577 
578         status = ret;
579         goto error;
580 
581     case NXT_AGAIN:
582         status = nxt_h1p_header_buffer_test(task, h1p, c, r->conf->socket_conf);
583 
584         if (nxt_fast_path(status == NXT_OK)) {
585             c->read_state = &nxt_h1p_header_parse_state;
586 
587             nxt_conn_read(task->thread->engine, c);
588             return;
589         }
590 
591         break;
592 
593     case NXT_HTTP_PARSE_INVALID:
594         status = NXT_HTTP_BAD_REQUEST;
595         break;
596 
597     case NXT_HTTP_PARSE_UNSUPPORTED_VERSION:
598         status = NXT_HTTP_VERSION_NOT_SUPPORTED;
599         break;
600 
601     case NXT_HTTP_PARSE_TOO_LARGE_FIELD:
602         status = NXT_HTTP_REQUEST_HEADER_FIELDS_TOO_LARGE;
603         break;
604 
605     default:
606     case NXT_ERROR:
607         status = NXT_HTTP_INTERNAL_SERVER_ERROR;
608         break;
609     }
610 
611     (void) nxt_h1p_header_process(task, h1p, r);
612 
613 error:
614 
615     h1p->keepalive = 0;
616 
617     nxt_http_request_error(task, r, status);
618 }
619 
620 
621 static nxt_int_t
622 nxt_h1p_header_process(nxt_task_t *task, nxt_h1proto_t *h1p,
623     nxt_http_request_t *r)
624 {
625     u_char     *m;
626     nxt_int_t  ret;
627 
628     r->target.start = h1p->parser.target_start;
629     r->target.length = h1p->parser.target_end - h1p->parser.target_start;
630 
631     if (h1p->parser.version.ui64 != 0) {
632         r->version.start = h1p->parser.version.str;
633         r->version.length = sizeof(h1p->parser.version.str);
634     }
635 
636     r->method = &h1p->parser.method;
637     r->path = &h1p->parser.path;
638     r->args = &h1p->parser.args;
639 
640     r->fields = h1p->parser.fields;
641 
642     ret = nxt_http_fields_process(r->fields, &nxt_h1p_fields_hash, r);
643     if (nxt_slow_path(ret != NXT_OK)) {
644         return ret;
645     }
646 
647     if (h1p->connection_upgrade && h1p->upgrade_websocket) {
648         m = h1p->parser.method.start;
649 
650         if (nxt_slow_path(h1p->parser.method.length != 3
651                           || m[0] != 'G'
652                           || m[1] != 'E'
653                           || m[2] != 'T'))
654         {
655             nxt_log(task, NXT_LOG_INFO, "h1p upgrade: bad method");
656 
657             return NXT_HTTP_BAD_REQUEST;
658         }
659 
660         if (nxt_slow_path(h1p->parser.version.s.minor != '1')) {
661             nxt_log(task, NXT_LOG_INFO, "h1p upgrade: bad protocol version");
662 
663             return NXT_HTTP_BAD_REQUEST;
664         }
665 
666         if (nxt_slow_path(h1p->websocket_key == NULL)) {
667             nxt_log(task, NXT_LOG_INFO,
668                     "h1p upgrade: bad or absent websocket key");
669 
670             return NXT_HTTP_BAD_REQUEST;
671         }
672 
673         if (nxt_slow_path(h1p->websocket_version_ok == 0)) {
674             nxt_log(task, NXT_LOG_INFO,
675                     "h1p upgrade: bad or absent websocket version");
676 
677             return NXT_HTTP_UPGRADE_REQUIRED;
678         }
679 
680         r->websocket_handshake = 1;
681     }
682 
683     return ret;
684 }
685 
686 
687 static nxt_int_t
688 nxt_h1p_header_buffer_test(nxt_task_t *task, nxt_h1proto_t *h1p, nxt_conn_t *c,
689     nxt_socket_conf_t *skcf)
690 {
691     size_t     size, used;
692     nxt_buf_t  *in, *b;
693 
694     in = c->read;
695 
696     if (nxt_buf_mem_free_size(&in->mem) == 0) {
697         size = skcf->large_header_buffer_size;
698         used = nxt_buf_mem_used_size(&in->mem);
699 
700         if (size <= used || h1p->nbuffers >= skcf->large_header_buffers) {
701             return NXT_HTTP_REQUEST_HEADER_FIELDS_TOO_LARGE;
702         }
703 
704         b = nxt_buf_mem_alloc(c->mem_pool, size, 0);
705         if (nxt_slow_path(b == NULL)) {
706             return NXT_HTTP_INTERNAL_SERVER_ERROR;
707         }
708 
709         b->mem.free = nxt_cpymem(b->mem.pos, in->mem.pos, used);
710 
711         in->next = h1p->buffers;
712         h1p->buffers = in;
713         h1p->nbuffers++;
714 
715         c->read = b;
716     }
717 
718     return NXT_OK;
719 }
720 
721 
722 static nxt_int_t
723 nxt_h1p_connection(void *ctx, nxt_http_field_t *field, uintptr_t data)
724 {
725     nxt_http_request_t  *r;
726 
727     r = ctx;
728     field->hopbyhop = 1;
729 
730     if (field->value_length == 5 && nxt_memcmp(field->value, "close", 5) == 0) {
731         r->proto.h1->keepalive = 0;
732 
733     } else if (field->value_length == 7
734                && nxt_memcasecmp(field->value, "upgrade", 7) == 0)
735     {
736         r->proto.h1->connection_upgrade = 1;
737     }
738 
739     return NXT_OK;
740 }
741 
742 
743 static nxt_int_t
744 nxt_h1p_upgrade(void *ctx, nxt_http_field_t *field, uintptr_t data)
745 {
746     nxt_http_request_t  *r;
747 
748     r = ctx;
749 
750     if (field->value_length == 9
751         && nxt_memcasecmp(field->value, "websocket", 9) == 0)
752     {
753         r->proto.h1->upgrade_websocket = 1;
754     }
755 
756     return NXT_OK;
757 }
758 
759 
760 static nxt_int_t
761 nxt_h1p_websocket_key(void *ctx, nxt_http_field_t *field, uintptr_t data)
762 {
763     nxt_http_request_t  *r;
764 
765     r = ctx;
766 
767     if (field->value_length == 24) {
768         r->proto.h1->websocket_key = field;
769     }
770 
771     return NXT_OK;
772 }
773 
774 
775 static nxt_int_t
776 nxt_h1p_websocket_version(void *ctx, nxt_http_field_t *field, uintptr_t data)
777 {
778     nxt_http_request_t  *r;
779 
780     r = ctx;
781 
782     if (field->value_length == 2
783         && field->value[0] == '1' && field->value[1] == '3')
784     {
785         r->proto.h1->websocket_version_ok = 1;
786     }
787 
788     return NXT_OK;
789 }
790 
791 
792 static nxt_int_t
793 nxt_h1p_transfer_encoding(void *ctx, nxt_http_field_t *field, uintptr_t data)
794 {
795     nxt_http_te_t       te;
796     nxt_http_request_t  *r;
797 
798     r = ctx;
799     field->skip = 1;
800     field->hopbyhop = 1;
801 
802     if (field->value_length == 7
803         && nxt_memcmp(field->value, "chunked", 7) == 0)
804     {
805         te = NXT_HTTP_TE_CHUNKED;
806 
807     } else {
808         te = NXT_HTTP_TE_UNSUPPORTED;
809     }
810 
811     r->proto.h1->transfer_encoding = te;
812 
813     return NXT_OK;
814 }
815 
816 
817 static void
818 nxt_h1p_request_body_read(nxt_task_t *task, nxt_http_request_t *r)
819 {
820     size_t             size, body_length, body_buffer_size, body_rest;
821     ssize_t            res;
822     nxt_str_t          *tmp_path, tmp_name;
823     nxt_buf_t          *in, *b;
824     nxt_conn_t         *c;
825     nxt_h1proto_t      *h1p;
826     nxt_http_status_t  status;
827 
828     static const nxt_str_t tmp_name_pattern = nxt_string("/req-XXXXXXXX");
829 
830     h1p = r->proto.h1;
831 
832     nxt_debug(task, "h1p request body read %O te:%d",
833               r->content_length_n, h1p->transfer_encoding);
834 
835     switch (h1p->transfer_encoding) {
836 
837     case NXT_HTTP_TE_CHUNKED:
838         status = NXT_HTTP_LENGTH_REQUIRED;
839         goto error;
840 
841     case NXT_HTTP_TE_UNSUPPORTED:
842         status = NXT_HTTP_NOT_IMPLEMENTED;
843         goto error;
844 
845     default:
846     case NXT_HTTP_TE_NONE:
847         break;
848     }
849 
850     if (r->content_length_n == -1 || r->content_length_n == 0) {
851         goto ready;
852     }
853 
854     body_length = (size_t) r->content_length_n;
855 
856     body_buffer_size = nxt_min(r->conf->socket_conf->body_buffer_size,
857                                body_length);
858 
859     if (body_length > body_buffer_size) {
860         tmp_path = &r->conf->socket_conf->body_temp_path;
861 
862         tmp_name.length = tmp_path->length + tmp_name_pattern.length;
863 
864         b = nxt_buf_file_alloc(r->mem_pool,
865                                body_buffer_size + sizeof(nxt_file_t)
866                                + tmp_name.length + 1, 0);
867 
868     } else {
869         /* This initialization required for CentOS 6, gcc 4.4.7. */
870         tmp_path = NULL;
871         tmp_name.length = 0;
872 
873         b = nxt_buf_mem_alloc(r->mem_pool, body_buffer_size, 0);
874     }
875 
876     if (nxt_slow_path(b == NULL)) {
877         status = NXT_HTTP_INTERNAL_SERVER_ERROR;
878         goto error;
879     }
880 
881     r->body = b;
882 
883     if (body_length > body_buffer_size) {
884         tmp_name.start = nxt_pointer_to(b->mem.start, sizeof(nxt_file_t));
885 
886         memcpy(tmp_name.start, tmp_path->start, tmp_path->length);
887         memcpy(tmp_name.start + tmp_path->length, tmp_name_pattern.start,
888                tmp_name_pattern.length);
889         tmp_name.start[tmp_name.length] = '\0';
890 
891         b->file = (nxt_file_t *) b->mem.start;
892         nxt_memzero(b->file, sizeof(nxt_file_t));
893         b->file->fd = -1;
894         b->file->size = body_length;
895 
896         b->mem.start += sizeof(nxt_file_t) + tmp_name.length + 1;
897         b->mem.pos = b->mem.start;
898         b->mem.free = b->mem.start;
899 
900         b->file->fd = mkstemp((char *) tmp_name.start);
901         if (nxt_slow_path(b->file->fd == -1)) {
902             nxt_log(task, NXT_LOG_ERR, "mkstemp() failed %E", nxt_errno);
903 
904             status = NXT_HTTP_INTERNAL_SERVER_ERROR;
905             goto error;
906         }
907 
908         nxt_debug(task, "create body tmp file \"%V\", %d",
909                   &tmp_name, b->file->fd);
910 
911         unlink((char *) tmp_name.start);
912     }
913 
914     body_rest = body_length;
915 
916     in = h1p->conn->read;
917 
918     size = nxt_buf_mem_used_size(&in->mem);
919 
920     if (size != 0) {
921         size = nxt_min(size, body_length);
922 
923         if (nxt_buf_is_file(b)) {
924             res = nxt_fd_write(b->file->fd, in->mem.pos, size);
925             if (nxt_slow_path(res < (ssize_t) size)) {
926                 status = NXT_HTTP_INTERNAL_SERVER_ERROR;
927                 goto error;
928             }
929 
930             b->file_end += size;
931 
932         } else {
933             size = nxt_min(body_buffer_size, size);
934             b->mem.free = nxt_cpymem(b->mem.free, in->mem.pos, size);
935             body_buffer_size -= size;
936         }
937 
938         in->mem.pos += size;
939         body_rest -= size;
940     }
941 
942     nxt_debug(task, "h1p body rest: %uz", body_rest);
943 
944     if (body_rest != 0) {
945         in->next = h1p->buffers;
946         h1p->buffers = in;
947         h1p->nbuffers++;
948 
949         c = h1p->conn;
950         c->read = b;
951         c->read_state = &nxt_h1p_read_body_state;
952 
953         nxt_conn_read(task->thread->engine, c);
954         return;
955     }
956 
957     if (nxt_buf_is_file(b)) {
958         b->mem.start = NULL;
959         b->mem.end = NULL;
960         b->mem.pos = NULL;
961         b->mem.free = NULL;
962     }
963 
964 ready:
965 
966     r->state->ready_handler(task, r, NULL);
967 
968     return;
969 
970 error:
971 
972     h1p->keepalive = 0;
973 
974     nxt_http_request_error(task, r, status);
975 }
976 
977 
978 static const nxt_conn_state_t  nxt_h1p_read_body_state
979     nxt_aligned(64) =
980 {
981     .ready_handler = nxt_h1p_conn_request_body_read,
982     .close_handler = nxt_h1p_conn_request_error,
983     .error_handler = nxt_h1p_conn_request_error,
984 
985     .timer_handler = nxt_h1p_conn_request_timeout,
986     .timer_value = nxt_h1p_conn_request_timer_value,
987     .timer_data = offsetof(nxt_socket_conf_t, body_read_timeout),
988     .timer_autoreset = 1,
989 };
990 
991 
992 static void
993 nxt_h1p_conn_request_body_read(nxt_task_t *task, void *obj, void *data)
994 {
995     size_t              size, body_rest;
996     ssize_t             res;
997     nxt_buf_t           *b;
998     nxt_conn_t          *c;
999     nxt_h1proto_t       *h1p;
1000     nxt_http_request_t  *r;
1001     nxt_event_engine_t  *engine;
1002 
1003     c = obj;
1004     h1p = data;
1005 
1006     nxt_debug(task, "h1p conn request body read");
1007 
1008     r = h1p->request;
1009 
1010     engine = task->thread->engine;
1011 
1012     b = c->read;
1013 
1014     if (nxt_buf_is_file(b)) {
1015         body_rest = b->file->size - b->file_end;
1016 
1017         size = nxt_buf_mem_used_size(&b->mem);
1018         size = nxt_min(size, body_rest);
1019 
1020         res = nxt_fd_write(b->file->fd, b->mem.pos, size);
1021         if (nxt_slow_path(res < (ssize_t) size)) {
1022             nxt_h1p_request_error(task, h1p, r);
1023             return;
1024         }
1025 
1026         b->file_end += size;
1027         body_rest -= res;
1028 
1029         b->mem.pos += size;
1030 
1031         if (b->mem.pos == b->mem.free) {
1032             if (body_rest >= (size_t) nxt_buf_mem_size(&b->mem)) {
1033                 b->mem.free = b->mem.start;
1034 
1035             } else {
1036                 /* This required to avoid reading next request. */
1037                 b->mem.free = b->mem.end - body_rest;
1038             }
1039 
1040             b->mem.pos = b->mem.free;
1041         }
1042 
1043     } else {
1044         body_rest = nxt_buf_mem_free_size(&c->read->mem);
1045     }
1046 
1047     nxt_debug(task, "h1p body rest: %uz", body_rest);
1048 
1049     if (body_rest != 0) {
1050         nxt_conn_read(engine, c);
1051 
1052     } else {
1053         if (nxt_buf_is_file(b)) {
1054             b->mem.start = NULL;
1055             b->mem.end = NULL;
1056             b->mem.pos = NULL;
1057             b->mem.free = NULL;
1058         }
1059 
1060         c->read = NULL;
1061 
1062         r->state->ready_handler(task, r, NULL);
1063     }
1064 }
1065 
1066 
1067 static void
1068 nxt_h1p_request_local_addr(nxt_task_t *task, nxt_http_request_t *r)
1069 {
1070     r->local = nxt_conn_local_addr(task, r->proto.h1->conn);
1071 }
1072 
1073 
1074 #define NXT_HTTP_LAST_INFORMATIONAL                                           \
1075     (NXT_HTTP_CONTINUE + nxt_nitems(nxt_http_informational) - 1)
1076 
1077 static const nxt_str_t  nxt_http_informational[] = {
1078     nxt_string("HTTP/1.1 100 Continue\r\n"),
1079     nxt_string("HTTP/1.1 101 Switching Protocols\r\n"),
1080 };
1081 
1082 
1083 #define NXT_HTTP_LAST_SUCCESS                                                 \
1084     (NXT_HTTP_OK + nxt_nitems(nxt_http_success) - 1)
1085 
1086 static const nxt_str_t  nxt_http_success[] = {
1087     nxt_string("HTTP/1.1 200 OK\r\n"),
1088     nxt_string("HTTP/1.1 201 Created\r\n"),
1089     nxt_string("HTTP/1.1 202 Accepted\r\n"),
1090     nxt_string("HTTP/1.1 203 Non-Authoritative Information\r\n"),
1091     nxt_string("HTTP/1.1 204 No Content\r\n"),
1092     nxt_string("HTTP/1.1 205 Reset Content\r\n"),
1093     nxt_string("HTTP/1.1 206 Partial Content\r\n"),
1094 };
1095 
1096 
1097 #define NXT_HTTP_LAST_REDIRECTION                                             \
1098     (NXT_HTTP_MULTIPLE_CHOICES + nxt_nitems(nxt_http_redirection) - 1)
1099 
1100 static const nxt_str_t  nxt_http_redirection[] = {
1101     nxt_string("HTTP/1.1 300 Multiple Choices\r\n"),
1102     nxt_string("HTTP/1.1 301 Moved Permanently\r\n"),
1103     nxt_string("HTTP/1.1 302 Found\r\n"),
1104     nxt_string("HTTP/1.1 303 See Other\r\n"),
1105     nxt_string("HTTP/1.1 304 Not Modified\r\n"),
1106 };
1107 
1108 
1109 #define NXT_HTTP_LAST_CLIENT_ERROR                                            \
1110     (NXT_HTTP_BAD_REQUEST + nxt_nitems(nxt_http_client_error) - 1)
1111 
1112 static const nxt_str_t  nxt_http_client_error[] = {
1113     nxt_string("HTTP/1.1 400 Bad Request\r\n"),
1114     nxt_string("HTTP/1.1 401 Unauthorized\r\n"),
1115     nxt_string("HTTP/1.1 402 Payment Required\r\n"),
1116     nxt_string("HTTP/1.1 403 Forbidden\r\n"),
1117     nxt_string("HTTP/1.1 404 Not Found\r\n"),
1118     nxt_string("HTTP/1.1 405 Method Not Allowed\r\n"),
1119     nxt_string("HTTP/1.1 406 Not Acceptable\r\n"),
1120     nxt_string("HTTP/1.1 407 Proxy Authentication Required\r\n"),
1121     nxt_string("HTTP/1.1 408 Request Timeout\r\n"),
1122     nxt_string("HTTP/1.1 409 Conflict\r\n"),
1123     nxt_string("HTTP/1.1 410 Gone\r\n"),
1124     nxt_string("HTTP/1.1 411 Length Required\r\n"),
1125     nxt_string("HTTP/1.1 412 Precondition Failed\r\n"),
1126     nxt_string("HTTP/1.1 413 Payload Too Large\r\n"),
1127     nxt_string("HTTP/1.1 414 URI Too Long\r\n"),
1128     nxt_string("HTTP/1.1 415 Unsupported Media Type\r\n"),
1129     nxt_string("HTTP/1.1 416 Range Not Satisfiable\r\n"),
1130     nxt_string("HTTP/1.1 417 Expectation Failed\r\n"),
1131     nxt_string("HTTP/1.1 418\r\n"),
1132     nxt_string("HTTP/1.1 419\r\n"),
1133     nxt_string("HTTP/1.1 420\r\n"),
1134     nxt_string("HTTP/1.1 421\r\n"),
1135     nxt_string("HTTP/1.1 422\r\n"),
1136     nxt_string("HTTP/1.1 423\r\n"),
1137     nxt_string("HTTP/1.1 424\r\n"),
1138     nxt_string("HTTP/1.1 425\r\n"),
1139     nxt_string("HTTP/1.1 426 Upgrade Required\r\n"),
1140     nxt_string("HTTP/1.1 427\r\n"),
1141     nxt_string("HTTP/1.1 428\r\n"),
1142     nxt_string("HTTP/1.1 429\r\n"),
1143     nxt_string("HTTP/1.1 430\r\n"),
1144     nxt_string("HTTP/1.1 431 Request Header Fields Too Large\r\n"),
1145 };
1146 
1147 
1148 #define NXT_HTTP_LAST_NGINX_ERROR                                             \
1149     (NXT_HTTP_TO_HTTPS + nxt_nitems(nxt_http_nginx_error) - 1)
1150 
1151 static const nxt_str_t  nxt_http_nginx_error[] = {
1152     nxt_string("HTTP/1.1 400 "
1153                "The plain HTTP request was sent to HTTPS port\r\n"),
1154 };
1155 
1156 
1157 #define NXT_HTTP_LAST_SERVER_ERROR                                            \
1158     (NXT_HTTP_INTERNAL_SERVER_ERROR + nxt_nitems(nxt_http_server_error) - 1)
1159 
1160 static const nxt_str_t  nxt_http_server_error[] = {
1161     nxt_string("HTTP/1.1 500 Internal Server Error\r\n"),
1162     nxt_string("HTTP/1.1 501 Not Implemented\r\n"),
1163     nxt_string("HTTP/1.1 502 Bad Gateway\r\n"),
1164     nxt_string("HTTP/1.1 503 Service Unavailable\r\n"),
1165     nxt_string("HTTP/1.1 504 Gateway Timeout\r\n"),
1166     nxt_string("HTTP/1.1 505 HTTP Version Not Supported\r\n"),
1167 };
1168 
1169 
1170 #define UNKNOWN_STATUS_LENGTH  nxt_length("HTTP/1.1 65536\r\n")
1171 
1172 static void
1173 nxt_h1p_request_header_send(nxt_task_t *task, nxt_http_request_t *r,
1174     nxt_work_handler_t body_handler, void *data)
1175 {
1176     u_char              *p;
1177     size_t              size;
1178     nxt_buf_t           *header;
1179     nxt_str_t           unknown_status;
1180     nxt_int_t           conn;
1181     nxt_uint_t          n;
1182     nxt_bool_t          http11;
1183     nxt_conn_t          *c;
1184     nxt_h1proto_t       *h1p;
1185     const nxt_str_t     *status;
1186     nxt_http_field_t    *field;
1187     u_char              buf[UNKNOWN_STATUS_LENGTH];
1188 
1189     static const char   chunked[] = "Transfer-Encoding: chunked\r\n";
1190     static const char   websocket_version[] = "Sec-WebSocket-Version: 13\r\n";
1191 
1192     static const nxt_str_t  connection[3] = {
1193         nxt_string("Connection: close\r\n"),
1194         nxt_string("Connection: keep-alive\r\n"),
1195         nxt_string("Upgrade: websocket\r\n"
1196                    "Connection: Upgrade\r\n"
1197                    "Sec-WebSocket-Accept: "),
1198     };
1199 
1200     nxt_debug(task, "h1p request header send");
1201 
1202     r->header_sent = 1;
1203     h1p = r->proto.h1;
1204     n = r->status;
1205 
1206     if (n >= NXT_HTTP_CONTINUE && n <= NXT_HTTP_LAST_INFORMATIONAL) {
1207         status = &nxt_http_informational[n - NXT_HTTP_CONTINUE];
1208 
1209     } else if (n >= NXT_HTTP_OK && n <= NXT_HTTP_LAST_SUCCESS) {
1210         status = &nxt_http_success[n - NXT_HTTP_OK];
1211 
1212     } else if (n >= NXT_HTTP_MULTIPLE_CHOICES
1213                && n <= NXT_HTTP_LAST_REDIRECTION)
1214     {
1215         status = &nxt_http_redirection[n - NXT_HTTP_MULTIPLE_CHOICES];
1216 
1217     } else if (n >= NXT_HTTP_BAD_REQUEST && n <= NXT_HTTP_LAST_CLIENT_ERROR) {
1218         status = &nxt_http_client_error[n - NXT_HTTP_BAD_REQUEST];
1219 
1220     } else if (n >= NXT_HTTP_TO_HTTPS && n <= NXT_HTTP_LAST_NGINX_ERROR) {
1221         status = &nxt_http_nginx_error[n - NXT_HTTP_TO_HTTPS];
1222 
1223     } else if (n >= NXT_HTTP_INTERNAL_SERVER_ERROR
1224                && n <= NXT_HTTP_LAST_SERVER_ERROR)
1225     {
1226         status = &nxt_http_server_error[n - NXT_HTTP_INTERNAL_SERVER_ERROR];
1227 
1228     } else {
1229         p = nxt_sprintf(buf, buf + UNKNOWN_STATUS_LENGTH,
1230                         "HTTP/1.1 %03d\r\n", n);
1231 
1232         unknown_status.length = p - buf;
1233         unknown_status.start = buf;
1234         status = &unknown_status;
1235     }
1236 
1237     size = status->length;
1238     /* Trailing CRLF at the end of header. */
1239     size += nxt_length("\r\n");
1240 
1241     conn = -1;
1242 
1243     if (r->websocket_handshake && n == NXT_HTTP_SWITCHING_PROTOCOLS) {
1244         h1p->websocket = 1;
1245         h1p->keepalive = 0;
1246         conn = 2;
1247         size += NXT_WEBSOCKET_ACCEPT_SIZE + 2;
1248 
1249     } else {
1250         http11 = (h1p->parser.version.s.minor != '0');
1251 
1252         if (r->resp.content_length == NULL || r->resp.content_length->skip) {
1253 
1254             if (http11) {
1255                 if (n != NXT_HTTP_NOT_MODIFIED
1256                     && n != NXT_HTTP_NO_CONTENT
1257                     && body_handler != NULL
1258                     && !h1p->websocket)
1259                 {
1260                     h1p->chunked = 1;
1261                     size += nxt_length(chunked);
1262                     /* Trailing CRLF will be added by the first chunk header. */
1263                     size -= nxt_length("\r\n");
1264                 }
1265 
1266             } else {
1267                 h1p->keepalive = 0;
1268             }
1269         }
1270 
1271         if (http11 ^ h1p->keepalive) {
1272             conn = h1p->keepalive;
1273         }
1274     }
1275 
1276     if (conn >= 0) {
1277         size += connection[conn].length;
1278     }
1279 
1280     nxt_list_each(field, r->resp.fields) {
1281 
1282         if (!field->skip) {
1283             size += field->name_length + field->value_length;
1284             size += nxt_length(": \r\n");
1285         }
1286 
1287     } nxt_list_loop;
1288 
1289     if (nxt_slow_path(n == NXT_HTTP_UPGRADE_REQUIRED)) {
1290         size += nxt_length(websocket_version);
1291     }
1292 
1293     header = nxt_http_buf_mem(task, r, size);
1294     if (nxt_slow_path(header == NULL)) {
1295         nxt_h1p_request_error(task, h1p, r);
1296         return;
1297     }
1298 
1299     p = nxt_cpymem(header->mem.free, status->start, status->length);
1300 
1301     nxt_list_each(field, r->resp.fields) {
1302 
1303         if (!field->skip) {
1304             p = nxt_cpymem(p, field->name, field->name_length);
1305             *p++ = ':'; *p++ = ' ';
1306             p = nxt_cpymem(p, field->value, field->value_length);
1307             *p++ = '\r'; *p++ = '\n';
1308         }
1309 
1310     } nxt_list_loop;
1311 
1312     if (conn >= 0) {
1313         p = nxt_cpymem(p, connection[conn].start, connection[conn].length);
1314     }
1315 
1316     if (h1p->websocket) {
1317         nxt_websocket_accept(p, h1p->websocket_key->value);
1318         p += NXT_WEBSOCKET_ACCEPT_SIZE;
1319 
1320         *p++ = '\r'; *p++ = '\n';
1321     }
1322 
1323     if (nxt_slow_path(n == NXT_HTTP_UPGRADE_REQUIRED)) {
1324         p = nxt_cpymem(p, websocket_version, nxt_length(websocket_version));
1325     }
1326 
1327     if (h1p->chunked) {
1328         p = nxt_cpymem(p, chunked, nxt_length(chunked));
1329         /* Trailing CRLF will be added by the first chunk header. */
1330 
1331     } else {
1332         *p++ = '\r'; *p++ = '\n';
1333     }
1334 
1335     header->mem.free = p;
1336 
1337     h1p->header_size = nxt_buf_mem_used_size(&header->mem);
1338 
1339     c = h1p->conn;
1340 
1341     c->write = header;
1342     h1p->conn_write_tail = &header->next;
1343     c->write_state = &nxt_h1p_request_send_state;
1344 
1345     if (body_handler != NULL) {
1346         /*
1347          * The body handler will run before c->io->write() handler,
1348          * because the latter was inqueued by nxt_conn_write()
1349          * in engine->write_work_queue.
1350          */
1351         nxt_work_queue_add(&task->thread->engine->fast_work_queue,
1352                            body_handler, task, r, data);
1353 
1354     } else {
1355         header->next = nxt_http_buf_last(r);
1356     }
1357 
1358     nxt_conn_write(task->thread->engine, c);
1359 
1360     if (h1p->websocket) {
1361         nxt_h1p_websocket_first_frame_start(task, r, c->read);
1362     }
1363 }
1364 
1365 
1366 void
1367 nxt_h1p_complete_buffers(nxt_task_t *task, nxt_h1proto_t *h1p, nxt_bool_t all)
1368 {
1369     size_t            size;
1370     nxt_buf_t         *b, *in, *next;
1371     nxt_conn_t        *c;
1372 
1373     nxt_debug(task, "h1p complete buffers");
1374 
1375     b = h1p->buffers;
1376     c = h1p->conn;
1377     in = c->read;
1378 
1379     if (b != NULL) {
1380         if (in == NULL) {
1381             /* A request with large body. */
1382             in = b;
1383             c->read = in;
1384 
1385             b = in->next;
1386             in->next = NULL;
1387         }
1388 
1389         while (b != NULL) {
1390             next = b->next;
1391             b->next = NULL;
1392 
1393             b->completion_handler(task, b, b->parent);
1394 
1395             b = next;
1396         }
1397 
1398         h1p->buffers = NULL;
1399         h1p->nbuffers = 0;
1400     }
1401 
1402     if (in != NULL) {
1403         size = nxt_buf_mem_used_size(&in->mem);
1404 
1405         if (size == 0 || all) {
1406             in->completion_handler(task, in, in->parent);
1407 
1408             c->read = NULL;
1409         }
1410     }
1411 }
1412 
1413 
1414 static const nxt_conn_state_t  nxt_h1p_request_send_state
1415     nxt_aligned(64) =
1416 {
1417     .ready_handler = nxt_h1p_conn_sent,
1418     .error_handler = nxt_h1p_conn_request_error,
1419 
1420     .timer_handler = nxt_h1p_conn_request_send_timeout,
1421     .timer_value = nxt_h1p_conn_request_timer_value,
1422     .timer_data = offsetof(nxt_socket_conf_t, send_timeout),
1423     .timer_autoreset = 1,
1424 };
1425 
1426 
1427 static void
1428 nxt_h1p_request_send(nxt_task_t *task, nxt_http_request_t *r, nxt_buf_t *out)
1429 {
1430     nxt_conn_t     *c;
1431     nxt_h1proto_t  *h1p;
1432 
1433     nxt_debug(task, "h1p request send");
1434 
1435     h1p = r->proto.h1;
1436     c = h1p->conn;
1437 
1438     if (h1p->chunked) {
1439         out = nxt_h1p_chunk_create(task, r, out);
1440         if (nxt_slow_path(out == NULL)) {
1441             nxt_h1p_request_error(task, h1p, r);
1442             return;
1443         }
1444     }
1445 
1446     if (c->write == NULL) {
1447         c->write = out;
1448         c->write_state = &nxt_h1p_request_send_state;
1449 
1450         nxt_conn_write(task->thread->engine, c);
1451 
1452     } else {
1453         *h1p->conn_write_tail = out;
1454     }
1455 
1456     while (out->next != NULL) {
1457         out = out->next;
1458     }
1459 
1460     h1p->conn_write_tail = &out->next;
1461 }
1462 
1463 
1464 static nxt_buf_t *
1465 nxt_h1p_chunk_create(nxt_task_t *task, nxt_http_request_t *r, nxt_buf_t *out)
1466 {
1467     nxt_off_t          size;
1468     nxt_buf_t          *b, **prev, *header, *tail;
1469 
1470     const size_t       chunk_size = 2 * nxt_length("\r\n") + NXT_OFF_T_HEXLEN;
1471     static const char  tail_chunk[] = "\r\n0\r\n\r\n";
1472 
1473     size = 0;
1474     prev = &out;
1475 
1476     for (b = out; b != NULL; b = b->next) {
1477 
1478         if (nxt_buf_is_last(b)) {
1479             tail = nxt_http_buf_mem(task, r, sizeof(tail_chunk));
1480             if (nxt_slow_path(tail == NULL)) {
1481                 return NULL;
1482             }
1483 
1484             *prev = tail;
1485             tail->next = b;
1486             /*
1487              * The tail_chunk size with trailing zero is 8 bytes, so
1488              * memcpy may be inlined with just single 8 byte move operation.
1489              */
1490             nxt_memcpy(tail->mem.free, tail_chunk, sizeof(tail_chunk));
1491             tail->mem.free += nxt_length(tail_chunk);
1492 
1493             break;
1494         }
1495 
1496         size += nxt_buf_used_size(b);
1497         prev = &b->next;
1498     }
1499 
1500     if (size == 0) {
1501         return out;
1502     }
1503 
1504     header = nxt_http_buf_mem(task, r, chunk_size);
1505     if (nxt_slow_path(header == NULL)) {
1506         return NULL;
1507     }
1508 
1509     header->next = out;
1510     header->mem.free = nxt_sprintf(header->mem.free, header->mem.end,
1511                                    "\r\n%xO\r\n", size);
1512     return header;
1513 }
1514 
1515 
1516 static nxt_off_t
1517 nxt_h1p_request_body_bytes_sent(nxt_task_t *task, nxt_http_proto_t proto)
1518 {
1519     nxt_off_t      sent;
1520     nxt_h1proto_t  *h1p;
1521 
1522     h1p = proto.h1;
1523 
1524     sent = h1p->conn->sent - h1p->header_size;
1525 
1526     return (sent > 0) ? sent : 0;
1527 }
1528 
1529 
1530 static void
1531 nxt_h1p_request_discard(nxt_task_t *task, nxt_http_request_t *r,
1532     nxt_buf_t *last)
1533 {
1534     nxt_buf_t         *b;
1535     nxt_conn_t        *c;
1536     nxt_h1proto_t     *h1p;
1537     nxt_work_queue_t  *wq;
1538 
1539     nxt_debug(task, "h1p request discard");
1540 
1541     h1p = r->proto.h1;
1542     h1p->keepalive = 0;
1543 
1544     c = h1p->conn;
1545     b = c->write;
1546     c->write = NULL;
1547 
1548     wq = &task->thread->engine->fast_work_queue;
1549 
1550     nxt_sendbuf_drain(task, wq, b);
1551     nxt_sendbuf_drain(task, wq, last);
1552 }
1553 
1554 
1555 static void
1556 nxt_h1p_conn_request_error(nxt_task_t *task, void *obj, void *data)
1557 {
1558     nxt_h1proto_t       *h1p;
1559     nxt_http_request_t  *r;
1560 
1561     h1p = data;
1562 
1563     nxt_debug(task, "h1p conn request error");
1564 
1565     r = h1p->request;
1566 
1567     if (nxt_slow_path(r == NULL)) {
1568         nxt_h1p_shutdown(task, h1p->conn);
1569         return;
1570     }
1571 
1572     if (r->fields == NULL) {
1573         (void) nxt_h1p_header_process(task, h1p, r);
1574     }
1575 
1576     if (r->status == 0) {
1577         r->status = NXT_HTTP_BAD_REQUEST;
1578     }
1579 
1580     nxt_h1p_request_error(task, h1p, r);
1581 }
1582 
1583 
1584 static void
1585 nxt_h1p_conn_request_timeout(nxt_task_t *task, void *obj, void *data)
1586 {
1587     nxt_conn_t          *c;
1588     nxt_timer_t         *timer;
1589     nxt_h1proto_t       *h1p;
1590     nxt_http_request_t  *r;
1591 
1592     timer = obj;
1593 
1594     nxt_debug(task, "h1p conn request timeout");
1595 
1596     c = nxt_read_timer_conn(timer);
1597     c->block_read = 1;
1598     /*
1599      * Disable SO_LINGER off during socket closing
1600      * to send "408 Request Timeout" error response.
1601      */
1602     c->socket.timedout = 0;
1603 
1604     h1p = c->socket.data;
1605     h1p->keepalive = 0;
1606     r = h1p->request;
1607 
1608     if (r->fields == NULL) {
1609         (void) nxt_h1p_header_process(task, h1p, r);
1610     }
1611 
1612     nxt_http_request_error(task, r, NXT_HTTP_REQUEST_TIMEOUT);
1613 }
1614 
1615 
1616 static void
1617 nxt_h1p_conn_request_send_timeout(nxt_task_t *task, void *obj, void *data)
1618 {
1619     nxt_conn_t     *c;
1620     nxt_timer_t    *timer;
1621     nxt_h1proto_t  *h1p;
1622 
1623     timer = obj;
1624 
1625     nxt_debug(task, "h1p conn request send timeout");
1626 
1627     c = nxt_write_timer_conn(timer);
1628     c->block_write = 1;
1629     h1p = c->socket.data;
1630 
1631     nxt_h1p_request_error(task, h1p, h1p->request);
1632 }
1633 
1634 
1635 nxt_msec_t
1636 nxt_h1p_conn_request_timer_value(nxt_conn_t *c, uintptr_t data)
1637 {
1638     nxt_h1proto_t  *h1p;
1639 
1640     h1p = c->socket.data;
1641 
1642     return nxt_value_at(nxt_msec_t, h1p->request->conf->socket_conf, data);
1643 }
1644 
1645 
1646 nxt_inline void
1647 nxt_h1p_request_error(nxt_task_t *task, nxt_h1proto_t *h1p,
1648     nxt_http_request_t *r)
1649 {
1650     h1p->keepalive = 0;
1651 
1652     r->state->error_handler(task, r, h1p);
1653 }
1654 
1655 
1656 static void
1657 nxt_h1p_request_close(nxt_task_t *task, nxt_http_proto_t proto,
1658     nxt_socket_conf_joint_t *joint)
1659 {
1660     nxt_conn_t     *c;
1661     nxt_h1proto_t  *h1p;
1662 
1663     nxt_debug(task, "h1p request close");
1664 
1665     h1p = proto.h1;
1666     h1p->keepalive &= !h1p->request->inconsistent;
1667     h1p->request = NULL;
1668 
1669     nxt_router_conf_release(task, joint);
1670 
1671     c = h1p->conn;
1672     task = &c->task;
1673     c->socket.task = task;
1674     c->read_timer.task = task;
1675     c->write_timer.task = task;
1676 
1677     if (h1p->keepalive) {
1678         nxt_h1p_keepalive(task, h1p, c);
1679 
1680     } else {
1681         nxt_h1p_shutdown(task, c);
1682     }
1683 }
1684 
1685 
1686 static void
1687 nxt_h1p_conn_sent(nxt_task_t *task, void *obj, void *data)
1688 {
1689     nxt_conn_t          *c;
1690     nxt_event_engine_t  *engine;
1691 
1692     c = obj;
1693 
1694     nxt_debug(task, "h1p conn sent");
1695 
1696     engine = task->thread->engine;
1697 
1698     c->write = nxt_sendbuf_completion(task, &engine->fast_work_queue, c->write);
1699 
1700     if (c->write != NULL) {
1701         nxt_conn_write(engine, c);
1702     }
1703 }
1704 
1705 
1706 static void
1707 nxt_h1p_conn_close(nxt_task_t *task, void *obj, void *data)
1708 {
1709     nxt_conn_t  *c;
1710 
1711     c = obj;
1712 
1713     nxt_debug(task, "h1p conn close");
1714 
1715     nxt_h1p_shutdown(task, c);
1716 }
1717 
1718 
1719 static void
1720 nxt_h1p_conn_error(nxt_task_t *task, void *obj, void *data)
1721 {
1722     nxt_conn_t  *c;
1723 
1724     c = obj;
1725 
1726     nxt_debug(task, "h1p conn error");
1727 
1728     nxt_h1p_shutdown(task, c);
1729 }
1730 
1731 
1732 static nxt_msec_t
1733 nxt_h1p_conn_timer_value(nxt_conn_t *c, uintptr_t data)
1734 {
1735     nxt_socket_conf_joint_t  *joint;
1736 
1737     joint = c->listen->socket.data;
1738 
1739     return nxt_value_at(nxt_msec_t, joint->socket_conf, data);
1740 }
1741 
1742 
1743 static void
1744 nxt_h1p_keepalive(nxt_task_t *task, nxt_h1proto_t *h1p, nxt_conn_t *c)
1745 {
1746     size_t     size;
1747     nxt_buf_t  *in;
1748 
1749     nxt_debug(task, "h1p keepalive");
1750 
1751     if (!c->tcp_nodelay) {
1752         nxt_conn_tcp_nodelay_on(task, c);
1753     }
1754 
1755     nxt_h1p_complete_buffers(task, h1p, 0);
1756 
1757     in = c->read;
1758 
1759     nxt_memzero(h1p, offsetof(nxt_h1proto_t, conn));
1760 
1761     c->sent = 0;
1762 
1763     if (in == NULL) {
1764         c->read_state = &nxt_h1p_keepalive_state;
1765 
1766         nxt_conn_read(task->thread->engine, c);
1767 
1768     } else {
1769         size = nxt_buf_mem_used_size(&in->mem);
1770 
1771         nxt_debug(task, "h1p pipelining");
1772 
1773         nxt_memmove(in->mem.start, in->mem.pos, size);
1774 
1775         in->mem.pos = in->mem.start;
1776         in->mem.free = in->mem.start + size;
1777 
1778         nxt_h1p_conn_request_init(task, c, c->socket.data);
1779     }
1780 }
1781 
1782 
1783 static const nxt_conn_state_t  nxt_h1p_keepalive_state
1784     nxt_aligned(64) =
1785 {
1786     .ready_handler = nxt_h1p_conn_request_init,
1787     .close_handler = nxt_h1p_conn_close,
1788     .error_handler = nxt_h1p_conn_error,
1789 
1790     .io_read_handler = nxt_h1p_idle_io_read_handler,
1791 
1792     .timer_handler = nxt_h1p_idle_timeout,
1793     .timer_value = nxt_h1p_conn_timer_value,
1794     .timer_data = offsetof(nxt_socket_conf_t, idle_timeout),
1795     .timer_autoreset = 1,
1796 };
1797 
1798 
1799 const nxt_conn_state_t  nxt_h1p_idle_close_state
1800     nxt_aligned(64) =
1801 {
1802     .close_handler = nxt_h1p_idle_close,
1803 };
1804 
1805 
1806 static void
1807 nxt_h1p_idle_close(nxt_task_t *task, void *obj, void *data)
1808 {
1809     nxt_conn_t  *c;
1810 
1811     c = obj;
1812 
1813     nxt_debug(task, "h1p idle close");
1814 
1815     nxt_h1p_idle_response(task, c);
1816 }
1817 
1818 
1819 static void
1820 nxt_h1p_idle_timeout(nxt_task_t *task, void *obj, void *data)
1821 {
1822     nxt_conn_t   *c;
1823     nxt_timer_t  *timer;
1824 
1825     timer = obj;
1826 
1827     nxt_debug(task, "h1p idle timeout");
1828 
1829     c = nxt_read_timer_conn(timer);
1830     c->block_read = 1;
1831 
1832     nxt_h1p_idle_response(task, c);
1833 }
1834 
1835 
1836 #define NXT_H1P_IDLE_TIMEOUT                                                  \
1837      "HTTP/1.1 408 Request Timeout\r\n"                                       \
1838      "Server: " NXT_SERVER "\r\n"                                             \
1839      "Connection: close\r\n"                                                  \
1840      "Content-Length: 0\r\n"                                                  \
1841      "Date: "
1842 
1843 
1844 static void
1845 nxt_h1p_idle_response(nxt_task_t *task, nxt_conn_t *c)
1846 {
1847     u_char         *p;
1848     size_t         size;
1849     nxt_buf_t      *out, *last;
1850     nxt_h1proto_t  *h1p;
1851 
1852     size = nxt_length(NXT_H1P_IDLE_TIMEOUT)
1853            + nxt_http_date_cache.size
1854            + nxt_length("\r\n\r\n");
1855 
1856     out = nxt_buf_mem_alloc(c->mem_pool, size, 0);
1857     if (nxt_slow_path(out == NULL)) {
1858         goto fail;
1859     }
1860 
1861     p = nxt_cpymem(out->mem.free, NXT_H1P_IDLE_TIMEOUT,
1862                    nxt_length(NXT_H1P_IDLE_TIMEOUT));
1863 
1864     p = nxt_thread_time_string(task->thread, &nxt_http_date_cache, p);
1865 
1866     out->mem.free = nxt_cpymem(p, "\r\n\r\n", 4);
1867 
1868     last = nxt_mp_zget(c->mem_pool, NXT_BUF_SYNC_SIZE);
1869     if (nxt_slow_path(last == NULL)) {
1870         goto fail;
1871     }
1872 
1873     out->next = last;
1874     nxt_buf_set_sync(last);
1875     nxt_buf_set_last(last);
1876 
1877     last->completion_handler = nxt_h1p_idle_response_sent;
1878     last->parent = c;
1879 
1880     h1p = c->socket.data;
1881     h1p->conn_write_tail = &last->next;
1882 
1883     c->write = out;
1884     c->write_state = &nxt_h1p_timeout_response_state;
1885 
1886     nxt_conn_write(task->thread->engine, c);
1887     return;
1888 
1889 fail:
1890 
1891     nxt_h1p_shutdown(task, c);
1892 }
1893 
1894 
1895 static const nxt_conn_state_t  nxt_h1p_timeout_response_state
1896     nxt_aligned(64) =
1897 {
1898     .ready_handler = nxt_h1p_conn_sent,
1899     .error_handler = nxt_h1p_conn_error,
1900 
1901     .timer_handler = nxt_h1p_idle_response_timeout,
1902     .timer_value = nxt_h1p_idle_response_timer_value,
1903 };
1904 
1905 
1906 static void
1907 nxt_h1p_idle_response_sent(nxt_task_t *task, void *obj, void *data)
1908 {
1909     nxt_conn_t  *c;
1910 
1911     c = data;
1912 
1913     nxt_debug(task, "h1p idle timeout response sent");
1914 
1915     nxt_h1p_shutdown(task, c);
1916 }
1917 
1918 
1919 static void
1920 nxt_h1p_idle_response_timeout(nxt_task_t *task, void *obj, void *data)
1921 {
1922     nxt_conn_t   *c;
1923     nxt_timer_t  *timer;
1924 
1925     timer = obj;
1926 
1927     nxt_debug(task, "h1p idle timeout response timeout");
1928 
1929     c = nxt_read_timer_conn(timer);
1930     c->block_write = 1;
1931 
1932     nxt_h1p_shutdown(task, c);
1933 }
1934 
1935 
1936 static nxt_msec_t
1937 nxt_h1p_idle_response_timer_value(nxt_conn_t *c, uintptr_t data)
1938 {
1939     return 10 * 1000;
1940 }
1941 
1942 
1943 static void
1944 nxt_h1p_shutdown(nxt_task_t *task, nxt_conn_t *c)
1945 {
1946     nxt_timer_t    *timer;
1947     nxt_h1proto_t  *h1p;
1948 
1949     nxt_debug(task, "h1p shutdown");
1950 
1951     h1p = c->socket.data;
1952 
1953     if (h1p != NULL) {
1954         nxt_h1p_complete_buffers(task, h1p, 1);
1955 
1956         if (nxt_slow_path(h1p->websocket_timer != NULL)) {
1957             timer = &h1p->websocket_timer->timer;
1958 
1959             if (timer->handler != nxt_h1p_conn_ws_shutdown) {
1960                 timer->handler = nxt_h1p_conn_ws_shutdown;
1961                 nxt_timer_add(task->thread->engine, timer, 0);
1962 
1963             } else {
1964                 nxt_debug(task, "h1p already scheduled ws shutdown");
1965             }
1966 
1967             return;
1968         }
1969     }
1970 
1971     nxt_h1p_closing(task, c);
1972 }
1973 
1974 
1975 static void
1976 nxt_h1p_conn_ws_shutdown(nxt_task_t *task, void *obj, void *data)
1977 {
1978     nxt_timer_t                *timer;
1979     nxt_h1p_websocket_timer_t  *ws_timer;
1980 
1981     nxt_debug(task, "h1p conn ws shutdown");
1982 
1983     timer = obj;
1984     ws_timer = nxt_timer_data(timer, nxt_h1p_websocket_timer_t, timer);
1985 
1986     nxt_h1p_closing(task, ws_timer->h1p->conn);
1987 }
1988 
1989 
1990 static void
1991 nxt_h1p_closing(nxt_task_t *task, nxt_conn_t *c)
1992 {
1993     nxt_debug(task, "h1p closing");
1994 
1995     c->socket.data = NULL;
1996 
1997 #if (NXT_TLS)
1998 
1999     if (c->u.tls != NULL) {
2000         c->write_state = &nxt_h1p_shutdown_state;
2001 
2002         c->io->shutdown(task, c, NULL);
2003         return;
2004     }
2005 
2006 #endif
2007 
2008     nxt_h1p_conn_closing(task, c, NULL);
2009 }
2010 
2011 
2012 #if (NXT_TLS)
2013 
2014 static const nxt_conn_state_t  nxt_h1p_shutdown_state
2015     nxt_aligned(64) =
2016 {
2017     .ready_handler = nxt_h1p_conn_closing,
2018     .close_handler = nxt_h1p_conn_closing,
2019     .error_handler = nxt_h1p_conn_closing,
2020 };
2021 
2022 #endif
2023 
2024 
2025 static void
2026 nxt_h1p_conn_closing(nxt_task_t *task, void *obj, void *data)
2027 {
2028     nxt_conn_t  *c;
2029 
2030     c = obj;
2031 
2032     nxt_debug(task, "h1p conn closing");
2033 
2034     c->write_state = &nxt_h1p_close_state;
2035 
2036     nxt_conn_close(task->thread->engine, c);
2037 }
2038 
2039 
2040 static const nxt_conn_state_t  nxt_h1p_close_state
2041     nxt_aligned(64) =
2042 {
2043     .ready_handler = nxt_h1p_conn_free,
2044 };
2045 
2046 
2047 static void
2048 nxt_h1p_conn_free(nxt_task_t *task, void *obj, void *data)
2049 {
2050     nxt_conn_t          *c;
2051     nxt_listen_event_t  *lev;
2052     nxt_event_engine_t  *engine;
2053 
2054     c = obj;
2055 
2056     nxt_debug(task, "h1p conn free");
2057 
2058     nxt_queue_remove(&c->link);
2059 
2060     engine = task->thread->engine;
2061 
2062     nxt_sockaddr_cache_free(engine, c);
2063 
2064     lev = c->listen;
2065 
2066     nxt_conn_free(task, c);
2067 
2068     nxt_router_listen_event_release(&engine->task, lev, NULL);
2069 }
2070 
2071 
2072 static void
2073 nxt_h1p_peer_connect(nxt_task_t *task, nxt_http_peer_t *peer)
2074 {
2075     nxt_mp_t            *mp;
2076     nxt_int_t           ret;
2077     nxt_conn_t          *c, *client;
2078     nxt_h1proto_t       *h1p;
2079     nxt_fd_event_t      *socket;
2080     nxt_work_queue_t    *wq;
2081     nxt_http_request_t  *r;
2082 
2083     nxt_debug(task, "h1p peer connect");
2084 
2085     peer->status = NXT_HTTP_UNSET;
2086     r = peer->request;
2087 
2088     mp = nxt_mp_create(1024, 128, 256, 32);
2089 
2090     if (nxt_slow_path(mp == NULL)) {
2091         goto fail;
2092     }
2093 
2094     h1p = nxt_mp_zalloc(mp, sizeof(nxt_h1proto_t));
2095     if (nxt_slow_path(h1p == NULL)) {
2096         goto fail;
2097     }
2098 
2099     ret = nxt_http_parse_request_init(&h1p->parser, r->mem_pool);
2100     if (nxt_slow_path(ret != NXT_OK)) {
2101         goto fail;
2102     }
2103 
2104     c = nxt_conn_create(mp, task);
2105     if (nxt_slow_path(c == NULL)) {
2106         goto fail;
2107     }
2108 
2109     c->mem_pool = mp;
2110     h1p->conn = c;
2111 
2112     peer->proto.h1 = h1p;
2113     h1p->request = r;
2114 
2115     c->socket.task = task;
2116     c->read_timer.task = task;
2117     c->write_timer.task = task;
2118     c->socket.data = peer;
2119     c->remote = peer->server->sockaddr;
2120 
2121     c->socket.write_ready = 1;
2122     c->write_state = &nxt_h1p_peer_connect_state;
2123 
2124     /*
2125      * TODO: queues should be implemented via client proto interface.
2126      */
2127     client = r->proto.h1->conn;
2128 
2129     socket = &client->socket;
2130     wq = socket->read_work_queue;
2131     c->read_work_queue = wq;
2132     c->socket.read_work_queue = wq;
2133     c->read_timer.work_queue = wq;
2134 
2135     wq = socket->write_work_queue;
2136     c->write_work_queue = wq;
2137     c->socket.write_work_queue = wq;
2138     c->write_timer.work_queue = wq;
2139     /* TODO END */
2140 
2141     nxt_conn_connect(task->thread->engine, c);
2142 
2143     return;
2144 
2145 fail:
2146 
2147     peer->status = NXT_HTTP_INTERNAL_SERVER_ERROR;
2148 
2149     r->state->error_handler(task, r, peer);
2150 }
2151 
2152 
2153 static const nxt_conn_state_t  nxt_h1p_peer_connect_state
2154     nxt_aligned(64) =
2155 {
2156     .ready_handler = nxt_h1p_peer_connected,
2157     .close_handler = nxt_h1p_peer_refused,
2158     .error_handler = nxt_h1p_peer_error,
2159 
2160     .timer_handler = nxt_h1p_peer_send_timeout,
2161     .timer_value = nxt_h1p_peer_timer_value,
2162     .timer_data = offsetof(nxt_socket_conf_t, proxy_timeout),
2163 };
2164 
2165 
2166 static void
2167 nxt_h1p_peer_connected(nxt_task_t *task, void *obj, void *data)
2168 {
2169     nxt_http_peer_t     *peer;
2170     nxt_http_request_t  *r;
2171 
2172     peer = data;
2173 
2174     nxt_debug(task, "h1p peer connected");
2175 
2176     r = peer->request;
2177     r->state->ready_handler(task, r, peer);
2178 }
2179 
2180 
2181 static void
2182 nxt_h1p_peer_refused(nxt_task_t *task, void *obj, void *data)
2183 {
2184     nxt_http_peer_t     *peer;
2185     nxt_http_request_t  *r;
2186 
2187     peer = data;
2188 
2189     nxt_debug(task, "h1p peer refused");
2190 
2191     //peer->status = NXT_HTTP_SERVICE_UNAVAILABLE;
2192     peer->status = NXT_HTTP_BAD_GATEWAY;
2193 
2194     r = peer->request;
2195     r->state->error_handler(task, r, peer);
2196 }
2197 
2198 
2199 static void
2200 nxt_h1p_peer_header_send(nxt_task_t *task, nxt_http_peer_t *peer)
2201 {
2202     u_char              *p;
2203     size_t              size;
2204     nxt_buf_t           *header, *body;
2205     nxt_conn_t          *c;
2206     nxt_http_field_t    *field;
2207     nxt_http_request_t  *r;
2208 
2209     nxt_debug(task, "h1p peer header send");
2210 
2211     r = peer->request;
2212 
2213     size = r->method->length + sizeof(" ") + r->target.length
2214            + sizeof(" HTTP/1.0\r\n")
2215            + sizeof("\r\n");
2216 
2217     nxt_list_each(field, r->fields) {
2218 
2219         if (!field->hopbyhop) {
2220             size += field->name_length + field->value_length;
2221             size += nxt_length(": \r\n");
2222         }
2223 
2224     } nxt_list_loop;
2225 
2226     header = nxt_http_buf_mem(task, r, size);
2227     if (nxt_slow_path(header == NULL)) {
2228         r->state->error_handler(task, r, peer);
2229         return;
2230     }
2231 
2232     p = header->mem.free;
2233 
2234     p = nxt_cpymem(p, r->method->start, r->method->length);
2235     *p++ = ' ';
2236     p = nxt_cpymem(p, r->target.start, r->target.length);
2237     p = nxt_cpymem(p, " HTTP/1.0\r\n", 11);
2238 
2239     nxt_list_each(field, r->fields) {
2240 
2241         if (!field->hopbyhop) {
2242             p = nxt_cpymem(p, field->name, field->name_length);
2243             *p++ = ':'; *p++ = ' ';
2244             p = nxt_cpymem(p, field->value, field->value_length);
2245             *p++ = '\r'; *p++ = '\n';
2246         }
2247 
2248     } nxt_list_loop;
2249 
2250     *p++ = '\r'; *p++ = '\n';
2251     header->mem.free = p;
2252     size = p - header->mem.pos;
2253 
2254     c = peer->proto.h1->conn;
2255     c->write = header;
2256     c->write_state = &nxt_h1p_peer_header_send_state;
2257 
2258     if (r->body != NULL) {
2259         if (nxt_buf_is_file(r->body)) {
2260             body = nxt_buf_file_alloc(r->mem_pool, 0, 0);
2261 
2262         } else {
2263             body = nxt_buf_mem_alloc(r->mem_pool, 0, 0);
2264         }
2265 
2266         if (nxt_slow_path(body == NULL)) {
2267             r->state->error_handler(task, r, peer);
2268             return;
2269         }
2270 
2271         header->next = body;
2272 
2273         if (nxt_buf_is_file(r->body)) {
2274             body->file = r->body->file;
2275             body->file_end = r->body->file_end;
2276 
2277         } else {
2278             body->mem = r->body->mem;
2279         }
2280 
2281         size += nxt_buf_used_size(body);
2282 
2283 //        nxt_mp_retain(r->mem_pool);
2284     }
2285 
2286     if (size > 16384) {
2287         /* Use proxy_send_timeout instead of proxy_timeout. */
2288         c->write_state = &nxt_h1p_peer_header_body_send_state;
2289     }
2290 
2291     nxt_conn_write(task->thread->engine, c);
2292 }
2293 
2294 
2295 static const nxt_conn_state_t  nxt_h1p_peer_header_send_state
2296     nxt_aligned(64) =
2297 {
2298     .ready_handler = nxt_h1p_peer_header_sent,
2299     .error_handler = nxt_h1p_peer_error,
2300 
2301     .timer_handler = nxt_h1p_peer_send_timeout,
2302     .timer_value = nxt_h1p_peer_timer_value,
2303     .timer_data = offsetof(nxt_socket_conf_t, proxy_timeout),
2304 };
2305 
2306 
2307 static const nxt_conn_state_t  nxt_h1p_peer_header_body_send_state
2308     nxt_aligned(64) =
2309 {
2310     .ready_handler = nxt_h1p_peer_header_sent,
2311     .error_handler = nxt_h1p_peer_error,
2312 
2313     .timer_handler = nxt_h1p_peer_send_timeout,
2314     .timer_value = nxt_h1p_peer_timer_value,
2315     .timer_data = offsetof(nxt_socket_conf_t, proxy_send_timeout),
2316     .timer_autoreset = 1,
2317 };
2318 
2319 
2320 static void
2321 nxt_h1p_peer_header_sent(nxt_task_t *task, void *obj, void *data)
2322 {
2323     nxt_conn_t          *c;
2324     nxt_http_peer_t     *peer;
2325     nxt_http_request_t  *r;
2326     nxt_event_engine_t  *engine;
2327 
2328     c = obj;
2329     peer = data;
2330 
2331     nxt_debug(task, "h1p peer header sent");
2332 
2333     engine = task->thread->engine;
2334 
2335     c->write = nxt_sendbuf_completion(task, &engine->fast_work_queue, c->write);
2336 
2337     if (c->write != NULL) {
2338         nxt_conn_write(engine, c);
2339         return;
2340     }
2341 
2342     r = peer->request;
2343     r->state->ready_handler(task, r, peer);
2344 }
2345 
2346 
2347 static void
2348 nxt_h1p_peer_header_read(nxt_task_t *task, nxt_http_peer_t *peer)
2349 {
2350     nxt_conn_t  *c;
2351 
2352     nxt_debug(task, "h1p peer header read");
2353 
2354     c = peer->proto.h1->conn;
2355 
2356     if (c->write_timer.enabled) {
2357         c->read_state = &nxt_h1p_peer_header_read_state;
2358 
2359     } else {
2360         c->read_state = &nxt_h1p_peer_header_read_timer_state;
2361     }
2362 
2363     nxt_conn_read(task->thread->engine, c);
2364 }
2365 
2366 
2367 static const nxt_conn_state_t  nxt_h1p_peer_header_read_state
2368     nxt_aligned(64) =
2369 {
2370     .ready_handler = nxt_h1p_peer_header_read_done,
2371     .close_handler = nxt_h1p_peer_closed,
2372     .error_handler = nxt_h1p_peer_error,
2373 
2374     .io_read_handler = nxt_h1p_peer_io_read_handler,
2375 };
2376 
2377 
2378 static const nxt_conn_state_t  nxt_h1p_peer_header_read_timer_state
2379     nxt_aligned(64) =
2380 {
2381     .ready_handler = nxt_h1p_peer_header_read_done,
2382     .close_handler = nxt_h1p_peer_closed,
2383     .error_handler = nxt_h1p_peer_error,
2384 
2385     .io_read_handler = nxt_h1p_peer_io_read_handler,
2386 
2387     .timer_handler = nxt_h1p_peer_read_timeout,
2388     .timer_value = nxt_h1p_peer_timer_value,
2389     .timer_data = offsetof(nxt_socket_conf_t, proxy_timeout),
2390 };
2391 
2392 
2393 static ssize_t
2394 nxt_h1p_peer_io_read_handler(nxt_task_t *task, nxt_conn_t *c)
2395 {
2396     size_t              size;
2397     ssize_t             n;
2398     nxt_buf_t           *b;
2399     nxt_http_peer_t     *peer;
2400     nxt_socket_conf_t   *skcf;
2401     nxt_http_request_t  *r;
2402 
2403     peer = c->socket.data;
2404     r = peer->request;
2405     b = c->read;
2406 
2407     if (b == NULL) {
2408         skcf = r->conf->socket_conf;
2409 
2410         size = (peer->header_received) ? skcf->proxy_buffer_size
2411                                        : skcf->proxy_header_buffer_size;
2412 
2413         nxt_debug(task, "h1p peer io read: %z", size);
2414 
2415         b = nxt_http_proxy_buf_mem_alloc(task, r, size);
2416         if (nxt_slow_path(b == NULL)) {
2417             c->socket.error = NXT_ENOMEM;
2418             return NXT_ERROR;
2419         }
2420     }
2421 
2422     n = c->io->recvbuf(c, b);
2423 
2424     if (n > 0) {
2425         c->read = b;
2426 
2427     } else {
2428         c->read = NULL;
2429         nxt_http_proxy_buf_mem_free(task, r, b);
2430     }
2431 
2432     return n;
2433 }
2434 
2435 
2436 static void
2437 nxt_h1p_peer_header_read_done(nxt_task_t *task, void *obj, void *data)
2438 {
2439     nxt_int_t           ret;
2440     nxt_buf_t           *b;
2441     nxt_conn_t          *c;
2442     nxt_http_peer_t     *peer;
2443     nxt_http_request_t  *r;
2444     nxt_event_engine_t  *engine;
2445 
2446     c = obj;
2447     peer = data;
2448 
2449     nxt_debug(task, "h1p peer header read done");
2450 
2451     b = c->read;
2452 
2453     ret = nxt_h1p_peer_header_parse(peer, &b->mem);
2454 
2455     r = peer->request;
2456 
2457     ret = nxt_expect(NXT_DONE, ret);
2458 
2459     if (ret != NXT_AGAIN) {
2460         engine = task->thread->engine;
2461         nxt_timer_disable(engine, &c->write_timer);
2462         nxt_timer_disable(engine, &c->read_timer);
2463     }
2464 
2465     switch (ret) {
2466 
2467     case NXT_DONE:
2468         peer->fields = peer->proto.h1->parser.fields;
2469 
2470         ret = nxt_http_fields_process(peer->fields,
2471                                       &nxt_h1p_peer_fields_hash, r);
2472         if (nxt_slow_path(ret != NXT_OK)) {
2473             peer->status = NXT_HTTP_INTERNAL_SERVER_ERROR;
2474             break;
2475         }
2476 
2477         c->read = NULL;
2478 
2479         if (nxt_buf_mem_used_size(&b->mem) != 0) {
2480             peer->body = b;
2481         }
2482 
2483         peer->header_received = 1;
2484 
2485         r->state->ready_handler(task, r, peer);
2486         return;
2487 
2488     case NXT_AGAIN:
2489         if (nxt_buf_mem_free_size(&b->mem) != 0) {
2490             nxt_conn_read(task->thread->engine, c);
2491             return;
2492         }
2493 
2494         /* Fall through. */
2495 
2496     default:
2497     case NXT_ERROR:
2498     case NXT_HTTP_PARSE_INVALID:
2499     case NXT_HTTP_PARSE_UNSUPPORTED_VERSION:
2500     case NXT_HTTP_PARSE_TOO_LARGE_FIELD:
2501         peer->status = NXT_HTTP_BAD_GATEWAY;
2502         break;
2503     }
2504 
2505     nxt_http_proxy_buf_mem_free(task, r, b);
2506 
2507     r->state->error_handler(task, r, peer);
2508 }
2509 
2510 
2511 static nxt_int_t
2512 nxt_h1p_peer_header_parse(nxt_http_peer_t *peer, nxt_buf_mem_t *bm)
2513 {
2514     u_char     *p;
2515     size_t     length;
2516     nxt_int_t  status;
2517 
2518     if (peer->status < 0) {
2519         length = nxt_buf_mem_used_size(bm);
2520 
2521         if (nxt_slow_path(length < 12)) {
2522             return NXT_AGAIN;
2523         }
2524 
2525         p = bm->pos;
2526 
2527         if (nxt_slow_path(nxt_memcmp(p, "HTTP/1.", 7) != 0
2528                           || (p[7] != '0' && p[7] != '1')))
2529         {
2530             return NXT_ERROR;
2531         }
2532 
2533         status = nxt_int_parse(&p[9], 3);
2534 
2535         if (nxt_slow_path(status < 0)) {
2536             return NXT_ERROR;
2537         }
2538 
2539         p += 12;
2540         length -= 12;
2541 
2542         p = nxt_memchr(p, '\n', length);
2543 
2544         if (nxt_slow_path(p == NULL)) {
2545             return NXT_AGAIN;
2546         }
2547 
2548         bm->pos = p + 1;
2549         peer->status = status;
2550     }
2551 
2552     return nxt_http_parse_fields(&peer->proto.h1->parser, bm);
2553 }
2554 
2555 
2556 static void
2557 nxt_h1p_peer_read(nxt_task_t *task, nxt_http_peer_t *peer)
2558 {
2559     nxt_conn_t  *c;
2560 
2561     nxt_debug(task, "h1p peer read");
2562 
2563     c = peer->proto.h1->conn;
2564     c->read_state = &nxt_h1p_peer_read_state;
2565 
2566     nxt_conn_read(task->thread->engine, c);
2567 }
2568 
2569 
2570 static const nxt_conn_state_t  nxt_h1p_peer_read_state
2571     nxt_aligned(64) =
2572 {
2573     .ready_handler = nxt_h1p_peer_read_done,
2574     .close_handler = nxt_h1p_peer_closed,
2575     .error_handler = nxt_h1p_peer_error,
2576 
2577     .io_read_handler = nxt_h1p_peer_io_read_handler,
2578 
2579     .timer_handler = nxt_h1p_peer_read_timeout,
2580     .timer_value = nxt_h1p_peer_timer_value,
2581     .timer_data = offsetof(nxt_socket_conf_t, proxy_read_timeout),
2582     .timer_autoreset = 1,
2583 };
2584 
2585 
2586 static void
2587 nxt_h1p_peer_read_done(nxt_task_t *task, void *obj, void *data)
2588 {
2589     nxt_conn_t          *c;
2590     nxt_http_peer_t     *peer;
2591     nxt_http_request_t  *r;
2592 
2593     c = obj;
2594     peer = data;
2595 
2596     nxt_debug(task, "h1p peer read done");
2597 
2598     peer->body = c->read;
2599     c->read = NULL;
2600 
2601     r = peer->request;
2602     r->state->ready_handler(task, r, peer);
2603 }
2604 
2605 
2606 static void
2607 nxt_h1p_peer_closed(nxt_task_t *task, void *obj, void *data)
2608 {
2609     nxt_http_peer_t     *peer;
2610     nxt_http_request_t  *r;
2611 
2612     peer = data;
2613 
2614     nxt_debug(task, "h1p peer closed");
2615 
2616     r = peer->request;
2617 
2618     if (peer->header_received) {
2619         peer->body = nxt_http_buf_last(r);
2620 
2621         peer->closed = 1;
2622 
2623         r->state->ready_handler(task, r, peer);
2624 
2625     } else {
2626         peer->status = NXT_HTTP_BAD_GATEWAY;
2627 
2628         r->state->error_handler(task, r, peer);
2629     }
2630 }
2631 
2632 
2633 static void
2634 nxt_h1p_peer_error(nxt_task_t *task, void *obj, void *data)
2635 {
2636     nxt_http_peer_t     *peer;
2637     nxt_http_request_t  *r;
2638 
2639     peer = data;
2640 
2641     nxt_debug(task, "h1p peer error");
2642 
2643     peer->status = NXT_HTTP_BAD_GATEWAY;
2644 
2645     r = peer->request;
2646     r->state->error_handler(task, r, peer);
2647 }
2648 
2649 
2650 static void
2651 nxt_h1p_peer_send_timeout(nxt_task_t *task, void *obj, void *data)
2652 {
2653     nxt_conn_t          *c;
2654     nxt_timer_t         *timer;
2655     nxt_http_peer_t     *peer;
2656     nxt_http_request_t  *r;
2657 
2658     timer = obj;
2659 
2660     nxt_debug(task, "h1p peer send timeout");
2661 
2662     c = nxt_write_timer_conn(timer);
2663     c->block_write = 1;
2664     c->block_read = 1;
2665 
2666     peer = c->socket.data;
2667     peer->status = NXT_HTTP_GATEWAY_TIMEOUT;
2668 
2669     r = peer->request;
2670     r->state->error_handler(task, r, peer);
2671 }
2672 
2673 
2674 static void
2675 nxt_h1p_peer_read_timeout(nxt_task_t *task, void *obj, void *data)
2676 {
2677     nxt_conn_t          *c;
2678     nxt_timer_t         *timer;
2679     nxt_http_peer_t     *peer;
2680     nxt_http_request_t  *r;
2681 
2682     timer = obj;
2683 
2684     nxt_debug(task, "h1p peer read timeout");
2685 
2686     c = nxt_read_timer_conn(timer);
2687     c->block_write = 1;
2688     c->block_read = 1;
2689 
2690     peer = c->socket.data;
2691     peer->status = NXT_HTTP_GATEWAY_TIMEOUT;
2692 
2693     r = peer->request;
2694     r->state->error_handler(task, r, peer);
2695 }
2696 
2697 
2698 static nxt_msec_t
2699 nxt_h1p_peer_timer_value(nxt_conn_t *c, uintptr_t data)
2700 {
2701     nxt_http_peer_t  *peer;
2702 
2703     peer = c->socket.data;
2704 
2705     return nxt_value_at(nxt_msec_t, peer->request->conf->socket_conf, data);
2706 }
2707 
2708 
2709 static void
2710 nxt_h1p_peer_close(nxt_task_t *task, nxt_http_peer_t *peer)
2711 {
2712     nxt_conn_t  *c;
2713 
2714     nxt_debug(task, "h1p peer close");
2715 
2716     peer->closed = 1;
2717 
2718     c = peer->proto.h1->conn;
2719     task = &c->task;
2720     c->socket.task = task;
2721     c->read_timer.task = task;
2722     c->write_timer.task = task;
2723 
2724     if (c->socket.fd != -1) {
2725         c->write_state = &nxt_h1p_peer_close_state;
2726 
2727         nxt_conn_close(task->thread->engine, c);
2728 
2729     } else {
2730         nxt_h1p_peer_free(task, c, NULL);
2731     }
2732 }
2733 
2734 
2735 static const nxt_conn_state_t  nxt_h1p_peer_close_state
2736     nxt_aligned(64) =
2737 {
2738     .ready_handler = nxt_h1p_peer_free,
2739 };
2740 
2741 
2742 static void
2743 nxt_h1p_peer_free(nxt_task_t *task, void *obj, void *data)
2744 {
2745     nxt_conn_t  *c;
2746 
2747     c = obj;
2748 
2749     nxt_debug(task, "h1p peer free");
2750 
2751     nxt_conn_free(task, c);
2752 }
2753