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