xref: /unit/src/ruby/nxt_ruby.c (revision 1691:a9d23e41c19b)
1 /*
2  * Copyright (C) Alexander Borisov
3  * Copyright (C) NGINX, Inc.
4  */
5 
6 #include <ruby/nxt_ruby.h>
7 
8 #include <nxt_unit.h>
9 #include <nxt_unit_request.h>
10 
11 #include <ruby/thread.h>
12 
13 #include NXT_RUBY_MOUNTS_H
14 
15 
16 #define NXT_RUBY_RACK_API_VERSION_MAJOR  1
17 #define NXT_RUBY_RACK_API_VERSION_MINOR  3
18 
19 
20 typedef struct {
21     nxt_task_t      *task;
22     nxt_str_t       *script;
23     nxt_ruby_ctx_t  *rctx;
24 } nxt_ruby_rack_init_t;
25 
26 
27 static nxt_int_t nxt_ruby_start(nxt_task_t *task,
28     nxt_process_data_t *data);
29 static VALUE nxt_ruby_init_basic(VALUE arg);
30 static VALUE nxt_ruby_rack_init(nxt_ruby_rack_init_t *rack_init);
31 
32 static VALUE nxt_ruby_require_rubygems(VALUE arg);
33 static VALUE nxt_ruby_bundler_setup(VALUE arg);
34 static VALUE nxt_ruby_require_rack(VALUE arg);
35 static VALUE nxt_ruby_rack_parse_script(VALUE ctx);
36 static VALUE nxt_ruby_rack_env_create(VALUE arg);
37 static int nxt_ruby_init_io(nxt_ruby_ctx_t *rctx);
38 static void nxt_ruby_request_handler(nxt_unit_request_info_t *req);
39 static void *nxt_ruby_request_handler_gvl(void *req);
40 static int nxt_ruby_ready_handler(nxt_unit_ctx_t *ctx);
41 static VALUE nxt_ruby_thread_func(VALUE arg);
42 static void *nxt_ruby_unit_run(void *ctx);
43 static void nxt_ruby_ubf(void *ctx);
44 static int nxt_ruby_init_threads(nxt_ruby_app_conf_t *c);
45 static void nxt_ruby_join_threads(nxt_unit_ctx_t *ctx,
46     nxt_ruby_app_conf_t *c);
47 
48 static VALUE nxt_ruby_rack_app_run(VALUE arg);
49 static int nxt_ruby_read_request(nxt_unit_request_info_t *req, VALUE hash_env);
50 nxt_inline void nxt_ruby_add_sptr(VALUE hash_env, VALUE name,
51     nxt_unit_sptr_t *sptr, uint32_t len);
52 static nxt_int_t nxt_ruby_rack_result_status(nxt_unit_request_info_t *req,
53     VALUE result);
54 static int nxt_ruby_rack_result_headers(nxt_unit_request_info_t *req,
55     VALUE result, nxt_int_t status);
56 static int nxt_ruby_hash_info(VALUE r_key, VALUE r_value, VALUE arg);
57 static int nxt_ruby_hash_add(VALUE r_key, VALUE r_value, VALUE arg);
58 static int nxt_ruby_rack_result_body(nxt_unit_request_info_t *req,
59     VALUE result);
60 static int nxt_ruby_rack_result_body_file_write(nxt_unit_request_info_t *req,
61     VALUE filepath);
62 static void *nxt_ruby_response_write_cb(void *read_info);
63 static VALUE nxt_ruby_rack_result_body_each(VALUE body, VALUE arg,
64     int argc, const VALUE *argv, VALUE blockarg);
65 static void *nxt_ruby_response_write(void *body);
66 
67 static void nxt_ruby_exception_log(nxt_unit_request_info_t *req,
68     uint32_t level, const char *desc);
69 
70 static void nxt_ruby_ctx_done(nxt_ruby_ctx_t *rctx);
71 static void nxt_ruby_atexit(void);
72 
73 
74 static uint32_t  compat[] = {
75     NXT_VERNUM, NXT_DEBUG,
76 };
77 
78 static VALUE  nxt_ruby_rackup;
79 static VALUE  nxt_ruby_call;
80 
81 static uint32_t        nxt_ruby_threads;
82 static nxt_ruby_ctx_t  *nxt_ruby_ctxs;
83 
84 NXT_EXPORT nxt_app_module_t  nxt_app_module = {
85     sizeof(compat),
86     compat,
87     nxt_string("ruby"),
88     ruby_version,
89     nxt_ruby_mounts,
90     nxt_nitems(nxt_ruby_mounts),
91     NULL,
92     nxt_ruby_start,
93 };
94 
95 typedef struct {
96     nxt_str_t  string;
97     VALUE      *v;
98 } nxt_ruby_string_t;
99 
100 static VALUE  nxt_rb_80_str;
101 static VALUE  nxt_rb_content_length_str;
102 static VALUE  nxt_rb_content_type_str;
103 static VALUE  nxt_rb_http_str;
104 static VALUE  nxt_rb_https_str;
105 static VALUE  nxt_rb_path_info_str;
106 static VALUE  nxt_rb_query_string_str;
107 static VALUE  nxt_rb_rack_url_scheme_str;
108 static VALUE  nxt_rb_remote_addr_str;
109 static VALUE  nxt_rb_request_method_str;
110 static VALUE  nxt_rb_request_uri_str;
111 static VALUE  nxt_rb_server_addr_str;
112 static VALUE  nxt_rb_server_name_str;
113 static VALUE  nxt_rb_server_port_str;
114 static VALUE  nxt_rb_server_protocol_str;
115 
116 static nxt_ruby_string_t nxt_rb_strings[] = {
117     { nxt_string("80"), &nxt_rb_80_str },
118     { nxt_string("CONTENT_LENGTH"), &nxt_rb_content_length_str },
119     { nxt_string("CONTENT_TYPE"), &nxt_rb_content_type_str },
120     { nxt_string("http"), &nxt_rb_http_str },
121     { nxt_string("https"), &nxt_rb_https_str },
122     { nxt_string("PATH_INFO"), &nxt_rb_path_info_str },
123     { nxt_string("QUERY_STRING"), &nxt_rb_query_string_str },
124     { nxt_string("rack.url_scheme"), &nxt_rb_rack_url_scheme_str },
125     { nxt_string("REMOTE_ADDR"), &nxt_rb_remote_addr_str },
126     { nxt_string("REQUEST_METHOD"), &nxt_rb_request_method_str },
127     { nxt_string("REQUEST_URI"), &nxt_rb_request_uri_str },
128     { nxt_string("SERVER_ADDR"), &nxt_rb_server_addr_str },
129     { nxt_string("SERVER_NAME"), &nxt_rb_server_name_str },
130     { nxt_string("SERVER_PORT"), &nxt_rb_server_port_str },
131     { nxt_string("SERVER_PROTOCOL"), &nxt_rb_server_protocol_str },
132     { nxt_null_string, NULL },
133 };
134 
135 
136 static int
137 nxt_ruby_init_strings(void)
138 {
139     VALUE              v;
140     nxt_ruby_string_t  *pstr;
141 
142     pstr = nxt_rb_strings;
143 
144     while (pstr->string.start != NULL) {
145         v = rb_str_new_static((char *) pstr->string.start, pstr->string.length);
146 
147         if (nxt_slow_path(v == Qnil)) {
148             nxt_unit_alert(NULL, "Ruby: failed to create const string '%.*s'",
149                            (int) pstr->string.length,
150                            (char *) pstr->string.start);
151 
152             return NXT_UNIT_ERROR;
153         }
154 
155         *pstr->v = v;
156 
157         rb_gc_register_address(pstr->v);
158 
159         pstr++;
160     }
161 
162     return NXT_UNIT_OK;
163 }
164 
165 
166 static void
167 nxt_ruby_done_strings(void)
168 {
169     nxt_ruby_string_t  *pstr;
170 
171     pstr = nxt_rb_strings;
172 
173     while (pstr->string.start != NULL) {
174         rb_gc_unregister_address(pstr->v);
175 
176         *pstr->v = Qnil;
177 
178         pstr++;
179     }
180 }
181 
182 
183 static nxt_int_t
184 nxt_ruby_start(nxt_task_t *task, nxt_process_data_t *data)
185 {
186     int                    state, rc;
187     VALUE                  res;
188     nxt_ruby_ctx_t         ruby_ctx;
189     nxt_unit_ctx_t         *unit_ctx;
190     nxt_unit_init_t        ruby_unit_init;
191     nxt_ruby_app_conf_t    *c;
192     nxt_ruby_rack_init_t   rack_init;
193     nxt_common_app_conf_t  *conf;
194 
195     static char  *argv[2] = { (char *) "NGINX_Unit", (char *) "-e0" };
196 
197     conf = data->app;
198     c = &conf->u.ruby;
199 
200     nxt_ruby_threads = c->threads;
201 
202     RUBY_INIT_STACK
203     ruby_init();
204     ruby_options(2, argv);
205     ruby_script("NGINX_Unit");
206 
207     ruby_ctx.env = Qnil;
208     ruby_ctx.io_input = Qnil;
209     ruby_ctx.io_error = Qnil;
210     ruby_ctx.thread = Qnil;
211     ruby_ctx.ctx = NULL;
212     ruby_ctx.req = NULL;
213 
214     rack_init.task = task;
215     rack_init.script = &c->script;
216     rack_init.rctx = &ruby_ctx;
217 
218     nxt_ruby_init_strings();
219 
220     res = rb_protect(nxt_ruby_init_basic,
221                      (VALUE) (uintptr_t) &rack_init, &state);
222     if (nxt_slow_path(res == Qnil || state != 0)) {
223         nxt_ruby_exception_log(NULL, NXT_LOG_ALERT,
224                                "Failed to init basic variables");
225         return NXT_ERROR;
226     }
227 
228     nxt_ruby_call = Qnil;
229 
230     nxt_ruby_rackup = nxt_ruby_rack_init(&rack_init);
231     if (nxt_slow_path(nxt_ruby_rackup == Qnil)) {
232         return NXT_ERROR;
233     }
234 
235     rb_gc_register_address(&nxt_ruby_rackup);
236 
237     nxt_ruby_call = rb_intern("call");
238     if (nxt_slow_path(nxt_ruby_call == Qnil)) {
239         nxt_alert(task, "Ruby: Unable to find rack entry point");
240 
241         goto fail;
242     }
243 
244     rb_gc_register_address(&nxt_ruby_call);
245 
246     ruby_ctx.env = rb_protect(nxt_ruby_rack_env_create,
247                               (VALUE) (uintptr_t) &ruby_ctx, &state);
248     if (nxt_slow_path(ruby_ctx.env == Qnil || state != 0)) {
249         nxt_ruby_exception_log(NULL, NXT_LOG_ALERT,
250                                "Failed to create 'environ' variable");
251         goto fail;
252     }
253 
254     rc = nxt_ruby_init_threads(c);
255     if (nxt_slow_path(rc == NXT_UNIT_ERROR)) {
256         goto fail;
257     }
258 
259     nxt_unit_default_init(task, &ruby_unit_init);
260 
261     ruby_unit_init.callbacks.request_handler = nxt_ruby_request_handler;
262     ruby_unit_init.callbacks.ready_handler = nxt_ruby_ready_handler;
263     ruby_unit_init.shm_limit = conf->shm_limit;
264     ruby_unit_init.data = c;
265     ruby_unit_init.ctx_data = &ruby_ctx;
266 
267     unit_ctx = nxt_unit_init(&ruby_unit_init);
268     if (nxt_slow_path(unit_ctx == NULL)) {
269         goto fail;
270     }
271 
272     rc = (intptr_t) rb_thread_call_without_gvl(nxt_ruby_unit_run, unit_ctx,
273                                                nxt_ruby_ubf, unit_ctx);
274 
275     nxt_ruby_join_threads(unit_ctx, c);
276 
277     nxt_unit_done(unit_ctx);
278 
279     nxt_ruby_ctx_done(&ruby_ctx);
280 
281     nxt_ruby_atexit();
282 
283     exit(rc);
284 
285     return NXT_OK;
286 
287 fail:
288 
289     nxt_ruby_join_threads(NULL, c);
290 
291     nxt_ruby_ctx_done(&ruby_ctx);
292 
293     nxt_ruby_atexit();
294 
295     return NXT_ERROR;
296 }
297 
298 
299 static VALUE
300 nxt_ruby_init_basic(VALUE arg)
301 {
302     int                   state;
303     nxt_ruby_rack_init_t  *rack_init;
304 
305     rack_init = (nxt_ruby_rack_init_t *) (uintptr_t) arg;
306 
307     state = rb_enc_find_index("encdb");
308     if (nxt_slow_path(state == 0)) {
309         nxt_alert(rack_init->task,
310                   "Ruby: Failed to find encoding index 'encdb'");
311 
312         return Qnil;
313     }
314 
315     rb_funcall(rb_cObject, rb_intern("require"), 1,
316                rb_str_new2("enc/trans/transdb"));
317 
318     return arg;
319 }
320 
321 
322 static VALUE
323 nxt_ruby_rack_init(nxt_ruby_rack_init_t *rack_init)
324 {
325     int    state;
326     VALUE  rackup, err;
327 
328     rb_protect(nxt_ruby_require_rubygems, Qnil, &state);
329     if (nxt_slow_path(state != 0)) {
330         nxt_ruby_exception_log(NULL, NXT_LOG_ALERT,
331                                "Failed to require 'rubygems' package");
332         return Qnil;
333     }
334 
335     rb_protect(nxt_ruby_bundler_setup, Qnil, &state);
336     if (state != 0) {
337         err = rb_errinfo();
338 
339         if (rb_obj_is_kind_of(err, rb_eLoadError) == Qfalse) {
340             nxt_ruby_exception_log(NULL, NXT_LOG_ALERT,
341                                    "Failed to require 'bundler/setup' package");
342             return Qnil;
343         }
344 
345         rb_set_errinfo(Qnil);
346     }
347 
348     rb_protect(nxt_ruby_require_rack, Qnil, &state);
349     if (nxt_slow_path(state != 0)) {
350         nxt_ruby_exception_log(NULL, NXT_LOG_ALERT,
351                                "Failed to require 'rack' package");
352         return Qnil;
353     }
354 
355     rackup = rb_protect(nxt_ruby_rack_parse_script,
356                         (VALUE) (uintptr_t) rack_init, &state);
357     if (nxt_slow_path(TYPE(rackup) != T_ARRAY || state != 0)) {
358         nxt_ruby_exception_log(NULL, NXT_LOG_ALERT,
359                                "Failed to parse rack script");
360         return Qnil;
361     }
362 
363     if (nxt_slow_path(RARRAY_LEN(rackup) < 1)) {
364         nxt_alert(rack_init->task, "Ruby: Invalid rack config file");
365         return Qnil;
366     }
367 
368     return RARRAY_PTR(rackup)[0];
369 }
370 
371 
372 static VALUE
373 nxt_ruby_require_rubygems(VALUE arg)
374 {
375     return rb_funcall(rb_cObject, rb_intern("require"), 1,
376                       rb_str_new2("rubygems"));
377 }
378 
379 
380 static VALUE
381 nxt_ruby_bundler_setup(VALUE arg)
382 {
383     return rb_funcall(rb_cObject, rb_intern("require"), 1,
384                       rb_str_new2("bundler/setup"));
385 }
386 
387 
388 static VALUE
389 nxt_ruby_require_rack(VALUE arg)
390 {
391     return rb_funcall(rb_cObject, rb_intern("require"), 1, rb_str_new2("rack"));
392 }
393 
394 
395 static VALUE
396 nxt_ruby_rack_parse_script(VALUE ctx)
397 {
398     VALUE                 script, res, rack, builder;
399     nxt_ruby_rack_init_t  *rack_init;
400 
401     rack_init = (nxt_ruby_rack_init_t *) (uintptr_t) ctx;
402 
403     rack = rb_const_get(rb_cObject, rb_intern("Rack"));
404     builder = rb_const_get(rack, rb_intern("Builder"));
405 
406     script = rb_str_new((const char *) rack_init->script->start,
407                         (long) rack_init->script->length);
408 
409     res = rb_funcall(builder, rb_intern("parse_file"), 1, script);
410 
411     rb_str_free(script);
412 
413     return res;
414 }
415 
416 
417 static VALUE
418 nxt_ruby_rack_env_create(VALUE arg)
419 {
420     int             rc;
421     VALUE           hash_env, version;
422     nxt_ruby_ctx_t  *rctx;
423 
424     rctx = (nxt_ruby_ctx_t *) (uintptr_t) arg;
425 
426     rc = nxt_ruby_init_io(rctx);
427     if (nxt_slow_path(rc != NXT_UNIT_OK)) {
428         return Qnil;
429     }
430 
431     hash_env = rb_hash_new();
432 
433     rb_hash_aset(hash_env, rb_str_new2("SERVER_SOFTWARE"),
434                  rb_str_new((const char *) nxt_server.start,
435                             (long) nxt_server.length));
436 
437     version = rb_ary_new();
438 
439     rb_ary_push(version, UINT2NUM(NXT_RUBY_RACK_API_VERSION_MAJOR));
440     rb_ary_push(version, UINT2NUM(NXT_RUBY_RACK_API_VERSION_MINOR));
441 
442     rb_hash_aset(hash_env, rb_str_new2("rack.version"), version);
443     rb_hash_aset(hash_env, rb_str_new2("rack.input"), rctx->io_input);
444     rb_hash_aset(hash_env, rb_str_new2("rack.errors"), rctx->io_error);
445     rb_hash_aset(hash_env, rb_str_new2("rack.multithread"),
446                  nxt_ruby_threads > 1 ? Qtrue : Qfalse);
447     rb_hash_aset(hash_env, rb_str_new2("rack.multiprocess"), Qtrue);
448     rb_hash_aset(hash_env, rb_str_new2("rack.run_once"), Qfalse);
449     rb_hash_aset(hash_env, rb_str_new2("rack.hijack?"), Qfalse);
450     rb_hash_aset(hash_env, rb_str_new2("rack.hijack"), Qnil);
451     rb_hash_aset(hash_env, rb_str_new2("rack.hijack_io"), Qnil);
452 
453     rctx->env = hash_env;
454 
455     rb_gc_register_address(&rctx->env);
456 
457     return hash_env;
458 }
459 
460 
461 static int
462 nxt_ruby_init_io(nxt_ruby_ctx_t *rctx)
463 {
464     VALUE  io_input, io_error;
465 
466     io_input = nxt_ruby_stream_io_input_init();
467 
468     rctx->io_input = rb_funcall(io_input, rb_intern("new"), 1,
469                                    (VALUE) (uintptr_t) rctx);
470     if (nxt_slow_path(rctx->io_input == Qnil)) {
471         nxt_unit_alert(NULL,
472                        "Ruby: Failed to create environment 'rack.input' var");
473 
474         return NXT_UNIT_ERROR;
475     }
476 
477     rb_gc_register_address(&rctx->io_input);
478 
479     io_error = nxt_ruby_stream_io_error_init();
480 
481     rctx->io_error = rb_funcall(io_error, rb_intern("new"), 1,
482                                    (VALUE) (uintptr_t) rctx);
483     if (nxt_slow_path(rctx->io_error == Qnil)) {
484         nxt_unit_alert(NULL,
485                        "Ruby: Failed to create environment 'rack.error' var");
486 
487         return NXT_UNIT_ERROR;
488     }
489 
490     rb_gc_register_address(&rctx->io_error);
491 
492     return NXT_UNIT_OK;
493 }
494 
495 
496 static void
497 nxt_ruby_request_handler(nxt_unit_request_info_t *req)
498 {
499     (void) rb_thread_call_with_gvl(nxt_ruby_request_handler_gvl, req);
500 }
501 
502 
503 static void *
504 nxt_ruby_request_handler_gvl(void *data)
505 {
506     int                      state;
507     VALUE                    res;
508     nxt_ruby_ctx_t           *rctx;
509     nxt_unit_request_info_t  *req;
510 
511     req = data;
512 
513     rctx = req->ctx->data;
514     rctx->req = req;
515 
516     res = rb_protect(nxt_ruby_rack_app_run, (VALUE) (uintptr_t) req, &state);
517     if (nxt_slow_path(res == Qnil || state != 0)) {
518         nxt_ruby_exception_log(req, NXT_LOG_ERR,
519                                "Failed to run ruby script");
520 
521         nxt_unit_request_done(req, NXT_UNIT_ERROR);
522 
523     } else {
524         nxt_unit_request_done(req, NXT_UNIT_OK);
525     }
526 
527     rctx->req = NULL;
528 
529     return NULL;
530 }
531 
532 
533 static VALUE
534 nxt_ruby_rack_app_run(VALUE arg)
535 {
536     int                      rc;
537     VALUE                    env, result;
538     nxt_int_t                status;
539     nxt_ruby_ctx_t           *rctx;
540     nxt_unit_request_info_t  *req;
541 
542     req = (nxt_unit_request_info_t *) arg;
543 
544     rctx = req->ctx->data;
545 
546     env = rb_hash_dup(rctx->env);
547 
548     rc = nxt_ruby_read_request(req, env);
549     if (nxt_slow_path(rc != NXT_UNIT_OK)) {
550         nxt_unit_req_alert(req,
551                            "Ruby: Failed to process incoming request");
552 
553         goto fail;
554     }
555 
556     result = rb_funcall(nxt_ruby_rackup, nxt_ruby_call, 1, env);
557     if (nxt_slow_path(TYPE(result) != T_ARRAY)) {
558         nxt_unit_req_error(req,
559                            "Ruby: Invalid response format from application");
560 
561         goto fail;
562     }
563 
564     if (nxt_slow_path(RARRAY_LEN(result) != 3)) {
565         nxt_unit_req_error(req,
566                            "Ruby: Invalid response format from application. "
567                            "Need 3 entries [Status, Headers, Body]");
568 
569         goto fail;
570     }
571 
572     status = nxt_ruby_rack_result_status(req, result);
573     if (nxt_slow_path(status < 0)) {
574         nxt_unit_req_error(req,
575                            "Ruby: Invalid response status from application.");
576 
577         goto fail;
578     }
579 
580     rc = nxt_ruby_rack_result_headers(req, result, status);
581     if (nxt_slow_path(rc != NXT_UNIT_OK)) {
582         goto fail;
583     }
584 
585     rc = nxt_ruby_rack_result_body(req, result);
586     if (nxt_slow_path(rc != NXT_UNIT_OK)) {
587         goto fail;
588     }
589 
590     rb_hash_delete(env, rb_obj_id(env));
591 
592     return result;
593 
594 fail:
595 
596     rb_hash_delete(env, rb_obj_id(env));
597 
598     return Qnil;
599 }
600 
601 
602 static int
603 nxt_ruby_read_request(nxt_unit_request_info_t *req, VALUE hash_env)
604 {
605     VALUE               name;
606     uint32_t            i;
607     nxt_unit_field_t    *f;
608     nxt_unit_request_t  *r;
609 
610     r = req->request;
611 
612     nxt_ruby_add_sptr(hash_env, nxt_rb_request_method_str, &r->method,
613                       r->method_length);
614     nxt_ruby_add_sptr(hash_env, nxt_rb_request_uri_str, &r->target,
615                       r->target_length);
616     nxt_ruby_add_sptr(hash_env, nxt_rb_path_info_str, &r->path, r->path_length);
617     nxt_ruby_add_sptr(hash_env, nxt_rb_query_string_str, &r->query,
618                       r->query_length);
619     nxt_ruby_add_sptr(hash_env, nxt_rb_server_protocol_str, &r->version,
620                       r->version_length);
621     nxt_ruby_add_sptr(hash_env, nxt_rb_remote_addr_str, &r->remote,
622                       r->remote_length);
623     nxt_ruby_add_sptr(hash_env, nxt_rb_server_addr_str, &r->local,
624                       r->local_length);
625     nxt_ruby_add_sptr(hash_env, nxt_rb_server_name_str, &r->server_name,
626                       r->server_name_length);
627 
628     rb_hash_aset(hash_env, nxt_rb_server_port_str, nxt_rb_80_str);
629 
630     rb_hash_aset(hash_env, nxt_rb_rack_url_scheme_str,
631                  r->tls ? nxt_rb_https_str : nxt_rb_http_str);
632 
633     for (i = 0; i < r->fields_count; i++) {
634         f = r->fields + i;
635 
636         name = rb_str_new(nxt_unit_sptr_get(&f->name), f->name_length);
637 
638         nxt_ruby_add_sptr(hash_env, name, &f->value, f->value_length);
639     }
640 
641     if (r->content_length_field != NXT_UNIT_NONE_FIELD) {
642         f = r->fields + r->content_length_field;
643 
644         nxt_ruby_add_sptr(hash_env, nxt_rb_content_length_str,
645                           &f->value, f->value_length);
646     }
647 
648     if (r->content_type_field != NXT_UNIT_NONE_FIELD) {
649         f = r->fields + r->content_type_field;
650 
651         nxt_ruby_add_sptr(hash_env, nxt_rb_content_type_str,
652                           &f->value, f->value_length);
653     }
654 
655     return NXT_UNIT_OK;
656 }
657 
658 
659 nxt_inline void
660 nxt_ruby_add_sptr(VALUE hash_env, VALUE name,
661     nxt_unit_sptr_t *sptr, uint32_t len)
662 {
663     char  *str;
664 
665     str = nxt_unit_sptr_get(sptr);
666 
667     rb_hash_aset(hash_env, name, rb_str_new(str, len));
668 }
669 
670 
671 static nxt_int_t
672 nxt_ruby_rack_result_status(nxt_unit_request_info_t *req, VALUE result)
673 {
674     VALUE   status;
675 
676     status = rb_ary_entry(result, 0);
677 
678     if (TYPE(status) == T_FIXNUM) {
679         return FIX2INT(status);
680     }
681 
682     if (TYPE(status) == T_STRING) {
683         return nxt_int_parse((u_char *) RSTRING_PTR(status),
684                              RSTRING_LEN(status));
685     }
686 
687     nxt_unit_req_error(req, "Ruby: Invalid response 'status' "
688                        "format from application");
689 
690     return -2;
691 }
692 
693 
694 typedef struct {
695     int                      rc;
696     uint32_t                 fields;
697     uint32_t                 size;
698     nxt_unit_request_info_t  *req;
699 } nxt_ruby_headers_info_t;
700 
701 
702 static int
703 nxt_ruby_rack_result_headers(nxt_unit_request_info_t *req, VALUE result,
704     nxt_int_t status)
705 {
706     int                      rc;
707     VALUE                    headers;
708     nxt_ruby_headers_info_t  headers_info;
709 
710     headers = rb_ary_entry(result, 1);
711     if (nxt_slow_path(TYPE(headers) != T_HASH)) {
712         nxt_unit_req_error(req,
713                            "Ruby: Invalid response 'headers' format from "
714                            "application");
715 
716         return NXT_UNIT_ERROR;
717     }
718 
719     rc = NXT_UNIT_OK;
720 
721     headers_info.rc = NXT_UNIT_OK;
722     headers_info.fields = 0;
723     headers_info.size = 0;
724     headers_info.req = req;
725 
726     rb_hash_foreach(headers, nxt_ruby_hash_info,
727                     (VALUE) (uintptr_t) &headers_info);
728     if (nxt_slow_path(headers_info.rc != NXT_UNIT_OK)) {
729         return headers_info.rc;
730     }
731 
732     rc = nxt_unit_response_init(req, status,
733                                 headers_info.fields, headers_info.size);
734     if (nxt_slow_path(rc != NXT_UNIT_OK)) {
735         return rc;
736     }
737 
738     rb_hash_foreach(headers, nxt_ruby_hash_add,
739                     (VALUE) (uintptr_t) &headers_info);
740 
741     return rc;
742 }
743 
744 
745 static int
746 nxt_ruby_hash_info(VALUE r_key, VALUE r_value, VALUE arg)
747 {
748     const char               *value, *value_end, *pos;
749     nxt_ruby_headers_info_t  *headers_info;
750 
751     headers_info = (void *) (uintptr_t) arg;
752 
753     if (nxt_slow_path(TYPE(r_key) != T_STRING)) {
754         nxt_unit_req_error(headers_info->req,
755                            "Ruby: Wrong header entry 'key' from application");
756 
757         goto fail;
758     }
759 
760     if (nxt_slow_path(TYPE(r_value) != T_STRING)) {
761         nxt_unit_req_error(headers_info->req,
762                            "Ruby: Wrong header entry 'value' from application");
763 
764         goto fail;
765     }
766 
767     value = RSTRING_PTR(r_value);
768     value_end = value + RSTRING_LEN(r_value);
769 
770     pos = value;
771 
772     for ( ;; ) {
773         pos = strchr(pos, '\n');
774 
775         if (pos == NULL) {
776             break;
777         }
778 
779         headers_info->fields++;
780         headers_info->size += RSTRING_LEN(r_key) + (pos - value);
781 
782         pos++;
783         value = pos;
784     }
785 
786     if (value <= value_end) {
787         headers_info->fields++;
788         headers_info->size += RSTRING_LEN(r_key) + (value_end - value);
789     }
790 
791     return ST_CONTINUE;
792 
793 fail:
794 
795     headers_info->rc = NXT_UNIT_ERROR;
796 
797     return ST_STOP;
798 }
799 
800 
801 static int
802 nxt_ruby_hash_add(VALUE r_key, VALUE r_value, VALUE arg)
803 {
804     int                      *rc;
805     uint32_t                 key_len;
806     const char               *value, *value_end, *pos;
807     nxt_ruby_headers_info_t  *headers_info;
808 
809     headers_info = (void *) (uintptr_t) arg;
810     rc = &headers_info->rc;
811 
812     value = RSTRING_PTR(r_value);
813     value_end = value + RSTRING_LEN(r_value);
814 
815     key_len = RSTRING_LEN(r_key);
816 
817     pos = value;
818 
819     for ( ;; ) {
820         pos = strchr(pos, '\n');
821 
822         if (pos == NULL) {
823             break;
824         }
825 
826         *rc = nxt_unit_response_add_field(headers_info->req,
827                                           RSTRING_PTR(r_key), key_len,
828                                           value, pos - value);
829         if (nxt_slow_path(*rc != NXT_UNIT_OK)) {
830             goto fail;
831         }
832 
833         pos++;
834         value = pos;
835     }
836 
837     if (value <= value_end) {
838         *rc = nxt_unit_response_add_field(headers_info->req,
839                                           RSTRING_PTR(r_key), key_len,
840                                           value, value_end - value);
841         if (nxt_slow_path(*rc != NXT_UNIT_OK)) {
842             goto fail;
843         }
844     }
845 
846     return ST_CONTINUE;
847 
848 fail:
849 
850     *rc = NXT_UNIT_ERROR;
851 
852     return ST_STOP;
853 }
854 
855 
856 static int
857 nxt_ruby_rack_result_body(nxt_unit_request_info_t *req, VALUE result)
858 {
859     int    rc;
860     VALUE  fn, body;
861 
862     body = rb_ary_entry(result, 2);
863 
864     if (rb_respond_to(body, rb_intern("to_path"))) {
865 
866         fn = rb_funcall(body, rb_intern("to_path"), 0);
867         if (nxt_slow_path(TYPE(fn) != T_STRING)) {
868             nxt_unit_req_error(req,
869                                "Ruby: Failed to get 'body' file path from "
870                                "application");
871 
872             return NXT_UNIT_ERROR;
873         }
874 
875         rc = nxt_ruby_rack_result_body_file_write(req, fn);
876         if (nxt_slow_path(rc != NXT_UNIT_OK)) {
877             return rc;
878         }
879 
880     } else if (rb_respond_to(body, rb_intern("each"))) {
881         rb_block_call(body, rb_intern("each"), 0, 0,
882                       nxt_ruby_rack_result_body_each, (VALUE) (uintptr_t) req);
883 
884     } else {
885         nxt_unit_req_error(req,
886                            "Ruby: Invalid response 'body' format "
887                            "from application");
888 
889         return NXT_UNIT_ERROR;
890     }
891 
892     if (rb_respond_to(body, rb_intern("close"))) {
893         rb_funcall(body, rb_intern("close"), 0);
894     }
895 
896     return NXT_UNIT_OK;
897 }
898 
899 
900 typedef struct {
901     int    fd;
902     off_t  pos;
903     off_t  rest;
904 } nxt_ruby_rack_file_t;
905 
906 
907 static ssize_t
908 nxt_ruby_rack_file_read(nxt_unit_read_info_t *read_info, void *dst, size_t size)
909 {
910     ssize_t               res;
911     nxt_ruby_rack_file_t  *file;
912 
913     file = read_info->data;
914 
915     size = nxt_min(size, (size_t) file->rest);
916 
917     res = pread(file->fd, dst, size, file->pos);
918 
919     if (res >= 0) {
920         file->pos += res;
921         file->rest -= res;
922 
923         if (size > (size_t) res) {
924             file->rest = 0;
925         }
926     }
927 
928     read_info->eof = file->rest == 0;
929 
930     return res;
931 }
932 
933 
934 typedef struct {
935     nxt_unit_read_info_t     read_info;
936     nxt_unit_request_info_t  *req;
937 } nxt_ruby_read_info_t;
938 
939 
940 static int
941 nxt_ruby_rack_result_body_file_write(nxt_unit_request_info_t *req,
942     VALUE filepath)
943 {
944     int                   fd, rc;
945     struct stat           finfo;
946     nxt_ruby_rack_file_t  ruby_file;
947     nxt_ruby_read_info_t  ri;
948 
949     fd = open(RSTRING_PTR(filepath), O_RDONLY, 0);
950     if (nxt_slow_path(fd == -1)) {
951         nxt_unit_req_error(req,
952                            "Ruby: Failed to open content file \"%s\": %s (%d)",
953                            RSTRING_PTR(filepath), strerror(errno), errno);
954 
955         return NXT_UNIT_ERROR;
956     }
957 
958     rc = fstat(fd, &finfo);
959     if (nxt_slow_path(rc == -1)) {
960         nxt_unit_req_error(req,
961                            "Ruby: Content file fstat(\"%s\") failed: %s (%d)",
962                            RSTRING_PTR(filepath), strerror(errno), errno);
963 
964         close(fd);
965 
966         return NXT_UNIT_ERROR;
967     }
968 
969     ruby_file.fd = fd;
970     ruby_file.pos = 0;
971     ruby_file.rest = finfo.st_size;
972 
973     ri.read_info.read = nxt_ruby_rack_file_read;
974     ri.read_info.eof = ruby_file.rest == 0;
975     ri.read_info.buf_size = ruby_file.rest;
976     ri.read_info.data = &ruby_file;
977     ri.req = req;
978 
979     rc = (intptr_t) rb_thread_call_without_gvl(nxt_ruby_response_write_cb,
980                                                &ri,
981                                                nxt_ruby_ubf,
982                                                req->ctx);
983 
984     close(fd);
985 
986     return rc;
987 }
988 
989 
990 static void *
991 nxt_ruby_response_write_cb(void *data)
992 {
993     int                   rc;
994     nxt_ruby_read_info_t  *ri;
995 
996     ri = data;
997 
998     rc = nxt_unit_response_write_cb(ri->req, &ri->read_info);
999     if (nxt_slow_path(rc != NXT_UNIT_OK)) {
1000         nxt_unit_req_error(ri->req, "Ruby: Failed to write content file.");
1001     }
1002 
1003     return (void *) (intptr_t) rc;
1004 }
1005 
1006 
1007 typedef struct {
1008     VALUE                    body;
1009     nxt_unit_request_info_t  *req;
1010 } nxt_ruby_write_info_t;
1011 
1012 
1013 static VALUE
1014 nxt_ruby_rack_result_body_each(VALUE body, VALUE arg, int argc,
1015     const VALUE *argv, VALUE blockarg)
1016 {
1017     nxt_ruby_write_info_t  wi;
1018 
1019     if (TYPE(body) != T_STRING) {
1020         return Qnil;
1021     }
1022 
1023     wi.body = body;
1024     wi.req = (void *) (uintptr_t) arg;
1025 
1026     (void) rb_thread_call_without_gvl(nxt_ruby_response_write,
1027                                       (void *) (uintptr_t) &wi,
1028                                       nxt_ruby_ubf, wi.req->ctx);
1029 
1030     return Qnil;
1031 }
1032 
1033 
1034 static void *
1035 nxt_ruby_response_write(void *data)
1036 {
1037     int                    rc;
1038     nxt_ruby_write_info_t  *wi;
1039 
1040     wi = data;
1041 
1042     rc = nxt_unit_response_write(wi->req, RSTRING_PTR(wi->body),
1043                                  RSTRING_LEN(wi->body));
1044     if (nxt_slow_path(rc != NXT_UNIT_OK)) {
1045         nxt_unit_req_error(wi->req,
1046                            "Ruby: Failed to write 'body' from application");
1047     }
1048 
1049     return (void *) (intptr_t) rc;
1050 }
1051 
1052 
1053 static void
1054 nxt_ruby_exception_log(nxt_unit_request_info_t *req, uint32_t level,
1055     const char *desc)
1056 {
1057     int    i;
1058     VALUE  err, ary, eclass, msg;
1059 
1060     nxt_unit_req_log(req, level, "Ruby: %s", desc);
1061 
1062     err = rb_errinfo();
1063     if (nxt_slow_path(err == Qnil)) {
1064         return;
1065     }
1066 
1067     ary = rb_funcall(err, rb_intern("backtrace"), 0);
1068     if (nxt_slow_path(RARRAY_LEN(ary) == 0)) {
1069         return;
1070     }
1071 
1072     eclass = rb_class_name(rb_class_of(err));
1073     msg = rb_funcall(err, rb_intern("message"), 0);
1074 
1075     nxt_unit_req_log(req, level, "Ruby: %s: %s (%s)",
1076                      RSTRING_PTR(RARRAY_PTR(ary)[0]),
1077                      RSTRING_PTR(msg), RSTRING_PTR(eclass));
1078 
1079     for (i = 1; i < RARRAY_LEN(ary); i++) {
1080         nxt_unit_req_log(req, level, "from %s",
1081                          RSTRING_PTR(RARRAY_PTR(ary)[i]));
1082     }
1083 }
1084 
1085 
1086 static void
1087 nxt_ruby_ctx_done(nxt_ruby_ctx_t *rctx)
1088 {
1089     if (rctx->io_input != Qnil) {
1090         rb_gc_unregister_address(&rctx->io_input);
1091     }
1092 
1093     if (rctx->io_error != Qnil) {
1094         rb_gc_unregister_address(&rctx->io_error);
1095     }
1096 
1097     if (rctx->env != Qnil) {
1098         rb_gc_unregister_address(&rctx->env);
1099     }
1100 }
1101 
1102 
1103 static void
1104 nxt_ruby_atexit(void)
1105 {
1106     if (nxt_ruby_rackup != Qnil) {
1107         rb_gc_unregister_address(&nxt_ruby_rackup);
1108     }
1109 
1110     if (nxt_ruby_call != Qnil) {
1111         rb_gc_unregister_address(&nxt_ruby_call);
1112     }
1113 
1114     nxt_ruby_done_strings();
1115 
1116     ruby_cleanup(0);
1117 }
1118 
1119 
1120 static int
1121 nxt_ruby_ready_handler(nxt_unit_ctx_t *ctx)
1122 {
1123     VALUE                res;
1124     uint32_t             i;
1125     nxt_ruby_ctx_t       *rctx;
1126     nxt_ruby_app_conf_t  *c;
1127 
1128     /* Worker thread context. */
1129     if (!nxt_unit_is_main_ctx(ctx)) {
1130         return NXT_UNIT_OK;
1131     }
1132 
1133     c = ctx->unit->data;
1134 
1135     if (c->threads <= 1) {
1136         return NXT_UNIT_OK;
1137     }
1138 
1139     for (i = 0; i < c->threads - 1; i++) {
1140         rctx = &nxt_ruby_ctxs[i];
1141 
1142         rctx->ctx = ctx;
1143 
1144         res = rb_thread_create(RUBY_METHOD_FUNC(nxt_ruby_thread_func), rctx);
1145 
1146         if (nxt_fast_path(res != Qnil)) {
1147             nxt_unit_debug(ctx, "thread #%d created", (int) (i + 1));
1148 
1149             rctx->thread = res;
1150 
1151         } else {
1152             nxt_unit_alert(ctx, "thread #%d create failed", (int) (i + 1));
1153 
1154             return NXT_UNIT_ERROR;
1155         }
1156     }
1157 
1158     return NXT_UNIT_OK;
1159 }
1160 
1161 
1162 static VALUE
1163 nxt_ruby_thread_func(VALUE arg)
1164 {
1165     nxt_unit_ctx_t  *ctx;
1166     nxt_ruby_ctx_t  *rctx;
1167 
1168     rctx = (nxt_ruby_ctx_t *) (uintptr_t) arg;
1169 
1170     nxt_unit_debug(rctx->ctx, "worker thread start");
1171 
1172     ctx = nxt_unit_ctx_alloc(rctx->ctx, rctx);
1173     if (nxt_slow_path(ctx == NULL)) {
1174         goto fail;
1175     }
1176 
1177     (void) rb_thread_call_without_gvl(nxt_ruby_unit_run, ctx,
1178                                       nxt_ruby_ubf, ctx);
1179 
1180     nxt_unit_done(ctx);
1181 
1182 fail:
1183 
1184     nxt_unit_debug(NULL, "worker thread end");
1185 
1186     return Qnil;
1187 }
1188 
1189 
1190 static void *
1191 nxt_ruby_unit_run(void *ctx)
1192 {
1193     return (void *) (intptr_t) nxt_unit_run(ctx);
1194 }
1195 
1196 
1197 static void
1198 nxt_ruby_ubf(void *ctx)
1199 {
1200     nxt_unit_warn(ctx, "Ruby: UBF");
1201 }
1202 
1203 
1204 static int
1205 nxt_ruby_init_threads(nxt_ruby_app_conf_t *c)
1206 {
1207     int             state;
1208     uint32_t        i;
1209     nxt_ruby_ctx_t  *rctx;
1210 
1211     if (c->threads <= 1) {
1212         return NXT_UNIT_OK;
1213     }
1214 
1215     nxt_ruby_ctxs = nxt_unit_malloc(NULL, sizeof(nxt_ruby_ctx_t)
1216                                           * (c->threads - 1));
1217     if (nxt_slow_path(nxt_ruby_ctxs == NULL)) {
1218         nxt_unit_alert(NULL, "Failed to allocate run contexts array");
1219 
1220         return NXT_UNIT_ERROR;
1221     }
1222 
1223     for (i = 0; i < c->threads - 1; i++) {
1224         rctx = &nxt_ruby_ctxs[i];
1225 
1226         rctx->env = Qnil;
1227         rctx->io_input = Qnil;
1228         rctx->io_error = Qnil;
1229         rctx->thread = Qnil;
1230     }
1231 
1232     for (i = 0; i < c->threads - 1; i++) {
1233         rctx = &nxt_ruby_ctxs[i];
1234 
1235         rctx->env = rb_protect(nxt_ruby_rack_env_create,
1236                                (VALUE) (uintptr_t) rctx, &state);
1237         if (nxt_slow_path(rctx->env == Qnil || state != 0)) {
1238             nxt_ruby_exception_log(NULL, NXT_LOG_ALERT,
1239                                    "Failed to create 'environ' variable");
1240             return NXT_UNIT_ERROR;
1241         }
1242     }
1243 
1244     return NXT_UNIT_OK;
1245 }
1246 
1247 
1248 static void
1249 nxt_ruby_join_threads(nxt_unit_ctx_t *ctx, nxt_ruby_app_conf_t *c)
1250 {
1251     uint32_t        i;
1252     nxt_ruby_ctx_t  *rctx;
1253 
1254     if (nxt_ruby_ctxs == NULL) {
1255         return;
1256     }
1257 
1258     for (i = 0; i < c->threads - 1; i++) {
1259         rctx = &nxt_ruby_ctxs[i];
1260 
1261         if (rctx->thread != Qnil) {
1262             rb_funcall(rctx->thread, rb_intern("join"), 0);
1263 
1264             nxt_unit_debug(ctx, "thread #%d joined", (int) (i + 1));
1265 
1266         } else {
1267             nxt_unit_debug(ctx, "thread #%d not started", (int) (i + 1));
1268         }
1269     }
1270 
1271     for (i = 0; i < c->threads - 1; i++) {
1272         nxt_ruby_ctx_done(&nxt_ruby_ctxs[i]);
1273     }
1274 
1275     nxt_unit_free(ctx, nxt_ruby_ctxs);
1276 }
1277