xref: /unit/src/python/nxt_python.c (revision 1980:43553aa72111)
1 
2 /*
3  * Copyright (C) NGINX, Inc.
4  */
5 
6 
7 #include <Python.h>
8 
9 #include <nxt_main.h>
10 #include <nxt_router.h>
11 #include <nxt_unit.h>
12 
13 #include <python/nxt_python.h>
14 
15 #include NXT_PYTHON_MOUNTS_H
16 
17 
18 typedef struct {
19     pthread_t       thread;
20     nxt_unit_ctx_t  *ctx;
21     void            *ctx_data;
22 } nxt_py_thread_info_t;
23 
24 
25 static nxt_int_t nxt_python_start(nxt_task_t *task,
26     nxt_process_data_t *data);
27 static nxt_int_t nxt_python_set_target(nxt_task_t *task,
28     nxt_python_target_t *target, nxt_conf_value_t *conf);
29 static nxt_int_t nxt_python_set_path(nxt_task_t *task, nxt_conf_value_t *value);
30 static int nxt_python_init_threads(nxt_python_app_conf_t *c);
31 static int nxt_python_ready_handler(nxt_unit_ctx_t *ctx);
32 static void *nxt_python_thread_func(void *main_ctx);
33 static void nxt_python_join_threads(nxt_unit_ctx_t *ctx,
34     nxt_python_app_conf_t *c);
35 static void nxt_python_atexit(void);
36 
37 static uint32_t  compat[] = {
38     NXT_VERNUM, NXT_DEBUG,
39 };
40 
41 
42 NXT_EXPORT nxt_app_module_t  nxt_app_module = {
43     sizeof(compat),
44     compat,
45     nxt_string("python"),
46     PY_VERSION,
47     nxt_python_mounts,
48     nxt_nitems(nxt_python_mounts),
49     NULL,
50     nxt_python_start,
51 };
52 
53 static PyObject           *nxt_py_stderr_flush;
54 nxt_python_targets_t      *nxt_py_targets;
55 
56 #if PY_MAJOR_VERSION == 3
57 static wchar_t            *nxt_py_home;
58 #else
59 static char               *nxt_py_home;
60 #endif
61 
62 static pthread_attr_t        *nxt_py_thread_attr;
63 static nxt_py_thread_info_t  *nxt_py_threads;
64 static nxt_python_proto_t    nxt_py_proto;
65 
66 
67 static nxt_int_t
nxt_python_start(nxt_task_t * task,nxt_process_data_t * data)68 nxt_python_start(nxt_task_t *task, nxt_process_data_t *data)
69 {
70     int                    rc;
71     size_t                 len, size;
72     uint32_t               next;
73     PyObject               *obj, *module;
74     nxt_str_t              proto, probe_proto, name;
75     nxt_int_t              ret, n, i;
76     nxt_unit_ctx_t         *unit_ctx;
77     nxt_unit_init_t        python_init;
78     nxt_conf_value_t       *cv;
79     nxt_python_targets_t   *targets;
80     nxt_common_app_conf_t  *app_conf;
81     nxt_python_app_conf_t  *c;
82 #if PY_MAJOR_VERSION == 3
83     char                   *path;
84     nxt_int_t              pep405;
85 
86     static const char pyvenv[] = "/pyvenv.cfg";
87     static const char bin_python[] = "/bin/python";
88 #endif
89 
90     static const nxt_str_t  wsgi = nxt_string("wsgi");
91     static const nxt_str_t  asgi = nxt_string("asgi");
92 
93     app_conf = data->app;
94     c = &app_conf->u.python;
95 
96     if (c->home != NULL) {
97         len = nxt_strlen(c->home);
98 
99 #if PY_MAJOR_VERSION == 3
100 
101         path = nxt_malloc(len + sizeof(pyvenv));
102         if (nxt_slow_path(path == NULL)) {
103             nxt_alert(task, "Failed to allocate memory");
104             return NXT_ERROR;
105         }
106 
107         nxt_memcpy(path, c->home, len);
108         nxt_memcpy(path + len, pyvenv, sizeof(pyvenv));
109 
110         pep405 = (access(path, R_OK) == 0);
111 
112         nxt_free(path);
113 
114         if (pep405) {
115             size = (len + sizeof(bin_python)) * sizeof(wchar_t);
116 
117         } else {
118             size = (len + 1) * sizeof(wchar_t);
119         }
120 
121         nxt_py_home = nxt_malloc(size);
122         if (nxt_slow_path(nxt_py_home == NULL)) {
123             nxt_alert(task, "Failed to allocate memory");
124             return NXT_ERROR;
125         }
126 
127         if (pep405) {
128             mbstowcs(nxt_py_home, c->home, len);
129             mbstowcs(nxt_py_home + len, bin_python, sizeof(bin_python));
130             Py_SetProgramName(nxt_py_home);
131 
132         } else {
133             mbstowcs(nxt_py_home, c->home, len + 1);
134             Py_SetPythonHome(nxt_py_home);
135         }
136 
137 #else
138         nxt_py_home = nxt_malloc(len + 1);
139         if (nxt_slow_path(nxt_py_home == NULL)) {
140             nxt_alert(task, "Failed to allocate memory");
141             return NXT_ERROR;
142         }
143 
144         nxt_memcpy(nxt_py_home, c->home, len + 1);
145         Py_SetPythonHome(nxt_py_home);
146 #endif
147     }
148 
149     Py_InitializeEx(0);
150 
151 #if PY_VERSION_HEX < NXT_PYTHON_VER(3, 7)
152     if (c->threads > 1) {
153         PyEval_InitThreads();
154     }
155 #endif
156 
157     module = NULL;
158     obj = NULL;
159 
160     python_init.ctx_data = NULL;
161 
162     obj = PySys_GetObject((char *) "stderr");
163     if (nxt_slow_path(obj == NULL)) {
164         nxt_alert(task, "Python failed to get \"sys.stderr\" object");
165         goto fail;
166     }
167 
168     nxt_py_stderr_flush = PyObject_GetAttrString(obj, "flush");
169 
170     /* obj is a Borrowed reference. */
171     obj = NULL;
172 
173     if (nxt_slow_path(nxt_py_stderr_flush == NULL)) {
174         nxt_alert(task, "Python failed to get \"flush\" attribute of "
175                         "\"sys.stderr\" object");
176         goto fail;
177     }
178 
179     if (nxt_slow_path(nxt_python_set_path(task, c->path) != NXT_OK)) {
180         goto fail;
181     }
182 
183     obj = Py_BuildValue("[s]", "unit");
184     if (nxt_slow_path(obj == NULL)) {
185         nxt_alert(task, "Python failed to create the \"sys.argv\" list");
186         goto fail;
187     }
188 
189     if (nxt_slow_path(PySys_SetObject((char *) "argv", obj) != 0)) {
190         nxt_alert(task, "Python failed to set the \"sys.argv\" list");
191         goto fail;
192     }
193 
194     Py_CLEAR(obj);
195 
196     n = (c->targets != NULL ? nxt_conf_object_members_count(c->targets) : 1);
197 
198     size = sizeof(nxt_python_targets_t) + n * sizeof(nxt_python_target_t);
199 
200     targets = nxt_unit_malloc(NULL, size);
201     if (nxt_slow_path(targets == NULL)) {
202         nxt_alert(task, "Could not allocate targets");
203         goto fail;
204     }
205 
206     memset(targets, 0, size);
207 
208     targets->count = n;
209     nxt_py_targets = targets;
210 
211     if (c->targets != NULL) {
212         next = 0;
213 
214         for (i = 0; /* void */; i++) {
215             cv = nxt_conf_next_object_member(c->targets, &name, &next);
216             if (cv == NULL) {
217                 break;
218             }
219 
220             ret = nxt_python_set_target(task, &targets->target[i], cv);
221             if (nxt_slow_path(ret != NXT_OK)) {
222                 goto fail;
223             }
224         }
225 
226     } else {
227         ret = nxt_python_set_target(task, &targets->target[0], app_conf->self);
228         if (nxt_slow_path(ret != NXT_OK)) {
229             goto fail;
230         }
231     }
232 
233     nxt_unit_default_init(task, &python_init, data->app);
234 
235     python_init.data = c;
236     python_init.callbacks.ready_handler = nxt_python_ready_handler;
237 
238     proto = c->protocol;
239 
240     if (proto.length == 0) {
241         proto = nxt_python_asgi_check(targets->target[0].application)
242                 ? asgi : wsgi;
243 
244         for (i = 1; i < targets->count; i++) {
245             probe_proto = nxt_python_asgi_check(targets->target[i].application)
246                           ? asgi : wsgi;
247             if (probe_proto.start != proto.start) {
248                 nxt_alert(task, "A mix of ASGI & WSGI targets is forbidden, "
249                                 "specify protocol in config if incorrect");
250                 goto fail;
251             }
252         }
253     }
254 
255     if (nxt_strstr_eq(&proto, &asgi)) {
256         rc = nxt_python_asgi_init(&python_init, &nxt_py_proto);
257 
258     } else {
259         rc = nxt_python_wsgi_init(&python_init, &nxt_py_proto);
260     }
261 
262     if (nxt_slow_path(rc == NXT_UNIT_ERROR)) {
263         goto fail;
264     }
265 
266     rc = nxt_py_proto.ctx_data_alloc(&python_init.ctx_data, 1);
267     if (nxt_slow_path(rc != NXT_UNIT_OK)) {
268         goto fail;
269     }
270 
271     rc = nxt_python_init_threads(c);
272     if (nxt_slow_path(rc == NXT_UNIT_ERROR)) {
273         goto fail;
274     }
275 
276     if (nxt_py_proto.startup != NULL) {
277         if (nxt_py_proto.startup(python_init.ctx_data) != NXT_UNIT_OK) {
278             goto fail;
279         }
280     }
281 
282     unit_ctx = nxt_unit_init(&python_init);
283     if (nxt_slow_path(unit_ctx == NULL)) {
284         goto fail;
285     }
286 
287     rc = nxt_py_proto.run(unit_ctx);
288 
289     nxt_python_join_threads(unit_ctx, c);
290 
291     nxt_unit_done(unit_ctx);
292 
293     nxt_py_proto.ctx_data_free(python_init.ctx_data);
294 
295     nxt_python_atexit();
296 
297     exit(rc);
298 
299     return NXT_OK;
300 
301 fail:
302 
303     nxt_python_join_threads(NULL, c);
304 
305     if (python_init.ctx_data != NULL) {
306         nxt_py_proto.ctx_data_free(python_init.ctx_data);
307     }
308 
309     Py_XDECREF(obj);
310     Py_XDECREF(module);
311 
312     nxt_python_atexit();
313 
314     return NXT_ERROR;
315 }
316 
317 
318 static nxt_int_t
nxt_python_set_target(nxt_task_t * task,nxt_python_target_t * target,nxt_conf_value_t * conf)319 nxt_python_set_target(nxt_task_t *task, nxt_python_target_t *target,
320     nxt_conf_value_t *conf)
321 {
322     char              *callable, *module_name;
323     PyObject          *module, *obj;
324     nxt_str_t         str;
325     nxt_conf_value_t  *value;
326 
327     static nxt_str_t  module_str = nxt_string("module");
328     static nxt_str_t  callable_str = nxt_string("callable");
329 
330     module = obj = NULL;
331 
332     value = nxt_conf_get_object_member(conf, &module_str, NULL);
333     if (nxt_slow_path(value == NULL)) {
334         goto fail;
335     }
336 
337     nxt_conf_get_string(value, &str);
338 
339     module_name = nxt_alloca(str.length + 1);
340     nxt_memcpy(module_name, str.start, str.length);
341     module_name[str.length] = '\0';
342 
343     module = PyImport_ImportModule(module_name);
344     if (nxt_slow_path(module == NULL)) {
345         nxt_alert(task, "Python failed to import module \"%s\"", module_name);
346         nxt_python_print_exception();
347         goto fail;
348     }
349 
350     value = nxt_conf_get_object_member(conf, &callable_str, NULL);
351     if (value == NULL) {
352         callable = nxt_alloca(12);
353         nxt_memcpy(callable, "application", 12);
354 
355     } else {
356         nxt_conf_get_string(value, &str);
357 
358         callable = nxt_alloca(str.length + 1);
359         nxt_memcpy(callable, str.start, str.length);
360         callable[str.length] = '\0';
361     }
362 
363     obj = PyDict_GetItemString(PyModule_GetDict(module), callable);
364     if (nxt_slow_path(obj == NULL)) {
365         nxt_alert(task, "Python failed to get \"%s\" from module \"%s\"",
366                   callable, module_name);
367         goto fail;
368     }
369 
370     if (nxt_slow_path(PyCallable_Check(obj) == 0)) {
371         nxt_alert(task, "\"%s\" in module \"%s\" is not a callable object",
372                   callable, module_name);
373         goto fail;
374     }
375 
376     target->application = obj;
377     obj = NULL;
378 
379     Py_INCREF(target->application);
380     Py_CLEAR(module);
381 
382     return NXT_OK;
383 
384 fail:
385 
386     Py_XDECREF(obj);
387     Py_XDECREF(module);
388 
389     return NXT_ERROR;
390 }
391 
392 
393 static nxt_int_t
nxt_python_set_path(nxt_task_t * task,nxt_conf_value_t * value)394 nxt_python_set_path(nxt_task_t *task, nxt_conf_value_t *value)
395 {
396     int               ret;
397     PyObject          *path, *sys;
398     nxt_str_t         str;
399     nxt_uint_t        n;
400     nxt_conf_value_t  *array;
401 
402     if (value == NULL) {
403         return NXT_OK;
404     }
405 
406     sys = PySys_GetObject((char *) "path");
407     if (nxt_slow_path(sys == NULL)) {
408         nxt_alert(task, "Python failed to get \"sys.path\" list");
409         return NXT_ERROR;
410     }
411 
412     /* sys is a Borrowed reference. */
413 
414     if (nxt_conf_type(value) == NXT_CONF_STRING) {
415         n = 0;
416         goto value_is_string;
417     }
418 
419     /* NXT_CONF_ARRAY */
420     array = value;
421 
422     n = nxt_conf_array_elements_count(array);
423 
424     while (n != 0) {
425         n--;
426 
427         /*
428          * Insertion in front of existing paths starting from the last element
429          * to preserve original order while giving priority to the values
430          * specified in the "path" option.
431          */
432 
433         value = nxt_conf_get_array_element(array, n);
434 
435     value_is_string:
436 
437         nxt_conf_get_string(value, &str);
438 
439         path = PyString_FromStringAndSize((char *) str.start, str.length);
440         if (nxt_slow_path(path == NULL)) {
441             nxt_alert(task, "Python failed to create string object \"%V\"",
442                       &str);
443             return NXT_ERROR;
444         }
445 
446         ret = PyList_Insert(sys, 0, path);
447 
448         Py_DECREF(path);
449 
450         if (nxt_slow_path(ret != 0)) {
451             nxt_alert(task, "Python failed to insert \"%V\" into \"sys.path\"",
452                       &str);
453             return NXT_ERROR;
454         }
455     }
456 
457     return NXT_OK;
458 }
459 
460 
461 static int
nxt_python_init_threads(nxt_python_app_conf_t * c)462 nxt_python_init_threads(nxt_python_app_conf_t *c)
463 {
464     int                    res;
465     uint32_t               i;
466     nxt_py_thread_info_t   *ti;
467     static pthread_attr_t  attr;
468 
469     if (c->threads <= 1) {
470         return NXT_UNIT_OK;
471     }
472 
473     if (c->thread_stack_size > 0) {
474         res = pthread_attr_init(&attr);
475         if (nxt_slow_path(res != 0)) {
476             nxt_unit_alert(NULL, "thread attr init failed: %s (%d)",
477                            strerror(res), res);
478 
479             return NXT_UNIT_ERROR;
480         }
481 
482         res = pthread_attr_setstacksize(&attr, c->thread_stack_size);
483         if (nxt_slow_path(res != 0)) {
484             nxt_unit_alert(NULL, "thread attr set stack size failed: %s (%d)",
485                            strerror(res), res);
486 
487             return NXT_UNIT_ERROR;
488         }
489 
490         nxt_py_thread_attr = &attr;
491     }
492 
493     nxt_py_threads = nxt_unit_malloc(NULL, sizeof(nxt_py_thread_info_t)
494                                            * (c->threads - 1));
495     if (nxt_slow_path(nxt_py_threads == NULL)) {
496         nxt_unit_alert(NULL, "Failed to allocate thread info array");
497 
498         return NXT_UNIT_ERROR;
499     }
500 
501     memset(nxt_py_threads, 0, sizeof(nxt_py_thread_info_t) * (c->threads - 1));
502 
503     for (i = 0; i < c->threads - 1; i++) {
504         ti = &nxt_py_threads[i];
505 
506         res = nxt_py_proto.ctx_data_alloc(&ti->ctx_data, 0);
507         if (nxt_slow_path(res != NXT_UNIT_OK)) {
508             return NXT_UNIT_ERROR;
509         }
510     }
511 
512     return NXT_UNIT_OK;
513 }
514 
515 
516 static int
nxt_python_ready_handler(nxt_unit_ctx_t * ctx)517 nxt_python_ready_handler(nxt_unit_ctx_t *ctx)
518 {
519     int                    res;
520     uint32_t               i;
521     nxt_py_thread_info_t   *ti;
522     nxt_python_app_conf_t  *c;
523 
524     c = ctx->unit->data;
525 
526     if (c->threads <= 1) {
527         return NXT_UNIT_OK;
528     }
529 
530     for (i = 0; i < c->threads - 1; i++) {
531         ti = &nxt_py_threads[i];
532 
533         ti->ctx = ctx;
534 
535         res = pthread_create(&ti->thread, nxt_py_thread_attr,
536                              nxt_python_thread_func, ti);
537 
538         if (nxt_fast_path(res == 0)) {
539             nxt_unit_debug(ctx, "thread #%d created", (int) (i + 1));
540 
541         } else {
542             nxt_unit_alert(ctx, "thread #%d create failed: %s (%d)",
543                            (int) (i + 1), strerror(res), res);
544         }
545     }
546 
547     return NXT_UNIT_OK;
548 }
549 
550 
551 static void *
nxt_python_thread_func(void * data)552 nxt_python_thread_func(void *data)
553 {
554     nxt_unit_ctx_t        *ctx;
555     PyGILState_STATE      gstate;
556     nxt_py_thread_info_t  *ti;
557 
558     ti = data;
559 
560     nxt_unit_debug(ti->ctx, "worker thread #%d start",
561                    (int) (ti - nxt_py_threads + 1));
562 
563     gstate = PyGILState_Ensure();
564 
565     if (nxt_py_proto.startup != NULL) {
566         if (nxt_py_proto.startup(ti->ctx_data) != NXT_UNIT_OK) {
567             goto fail;
568         }
569     }
570 
571     ctx = nxt_unit_ctx_alloc(ti->ctx, ti->ctx_data);
572     if (nxt_slow_path(ctx == NULL)) {
573         goto fail;
574     }
575 
576     (void) nxt_py_proto.run(ctx);
577 
578     nxt_unit_done(ctx);
579 
580 fail:
581 
582     PyGILState_Release(gstate);
583 
584     nxt_unit_debug(NULL, "worker thread #%d end",
585                    (int) (ti - nxt_py_threads + 1));
586 
587     return NULL;
588 }
589 
590 
591 static void
nxt_python_join_threads(nxt_unit_ctx_t * ctx,nxt_python_app_conf_t * c)592 nxt_python_join_threads(nxt_unit_ctx_t *ctx, nxt_python_app_conf_t *c)
593 {
594     int                   res;
595     uint32_t              i;
596     PyThreadState         *thread_state;
597     nxt_py_thread_info_t  *ti;
598 
599     if (nxt_py_threads == NULL) {
600         return;
601     }
602 
603     thread_state = PyEval_SaveThread();
604 
605     for (i = 0; i < c->threads - 1; i++) {
606         ti = &nxt_py_threads[i];
607 
608         if ((uintptr_t) ti->thread == 0) {
609             continue;
610         }
611 
612         res = pthread_join(ti->thread, NULL);
613 
614         if (nxt_fast_path(res == 0)) {
615             nxt_unit_debug(ctx, "thread #%d joined", (int) (i + 1));
616 
617         } else {
618             nxt_unit_alert(ctx, "thread #%d join failed: %s (%d)",
619                            (int) (i + 1), strerror(res), res);
620         }
621     }
622 
623     PyEval_RestoreThread(thread_state);
624 
625     for (i = 0; i < c->threads - 1; i++) {
626         ti = &nxt_py_threads[i];
627 
628         if (ti->ctx_data != NULL) {
629             nxt_py_proto.ctx_data_free(ti->ctx_data);
630         }
631     }
632 
633     nxt_unit_free(NULL, nxt_py_threads);
634 }
635 
636 
637 int
nxt_python_init_strings(nxt_python_string_t * pstr)638 nxt_python_init_strings(nxt_python_string_t *pstr)
639 {
640     PyObject  *obj;
641 
642     while (pstr->string.start != NULL) {
643         obj = PyString_FromStringAndSize((char *) pstr->string.start,
644                                          pstr->string.length);
645         if (nxt_slow_path(obj == NULL)) {
646             return NXT_UNIT_ERROR;
647         }
648 
649         PyUnicode_InternInPlace(&obj);
650 
651         *pstr->object_p = obj;
652 
653         pstr++;
654     }
655 
656     return NXT_UNIT_OK;
657 }
658 
659 
660 void
nxt_python_done_strings(nxt_python_string_t * pstr)661 nxt_python_done_strings(nxt_python_string_t *pstr)
662 {
663     PyObject  *obj;
664 
665     while (pstr->string.start != NULL) {
666         obj = *pstr->object_p;
667 
668         Py_XDECREF(obj);
669         *pstr->object_p = NULL;
670 
671         pstr++;
672     }
673 }
674 
675 
676 static void
nxt_python_atexit(void)677 nxt_python_atexit(void)
678 {
679     nxt_int_t  i;
680 
681     if (nxt_py_proto.done != NULL) {
682         nxt_py_proto.done();
683     }
684 
685     Py_XDECREF(nxt_py_stderr_flush);
686 
687     if (nxt_py_targets != NULL) {
688         for (i = 0; i < nxt_py_targets->count; i++) {
689             Py_XDECREF(nxt_py_targets->target[i].application);
690         }
691 
692         nxt_unit_free(NULL, nxt_py_targets);
693     }
694 
695     Py_Finalize();
696 
697     if (nxt_py_home != NULL) {
698         nxt_free(nxt_py_home);
699     }
700 }
701 
702 
703 void
nxt_python_print_exception(void)704 nxt_python_print_exception(void)
705 {
706     PyErr_Print();
707 
708 #if PY_MAJOR_VERSION == 3
709     /* The backtrace may be buffered in sys.stderr file object. */
710     {
711         PyObject  *result;
712 
713         result = PyObject_CallFunction(nxt_py_stderr_flush, NULL);
714         if (nxt_slow_path(result == NULL)) {
715             PyErr_Clear();
716             return;
717         }
718 
719         Py_DECREF(result);
720     }
721 #endif
722 }
723