xref: /unit/src/python/nxt_python_asgi_lifespan.c (revision 1872:9f8df8b810e0)
1 
2 /*
3  * Copyright (C) NGINX, Inc.
4  */
5 
6 
7 #include <python/nxt_python.h>
8 
9 #if (NXT_HAVE_ASGI)
10 
11 #include <nxt_main.h>
12 #include <python/nxt_python_asgi.h>
13 #include <python/nxt_python_asgi_str.h>
14 
15 
16 typedef struct  {
17     PyObject_HEAD
18     nxt_py_asgi_ctx_data_t  *ctx_data;
19     int                     disabled;
20     int                     startup_received;
21     int                     startup_sent;
22     int                     shutdown_received;
23     int                     shutdown_sent;
24     int                     shutdown_called;
25     PyObject                *startup_future;
26     PyObject                *shutdown_future;
27     PyObject                *receive_future;
28 } nxt_py_asgi_lifespan_t;
29 
30 static PyObject *nxt_py_asgi_lifespan_target_startup(
31     nxt_py_asgi_ctx_data_t *ctx_data, nxt_python_target_t *target);
32 static int nxt_py_asgi_lifespan_target_shutdown(
33     nxt_py_asgi_lifespan_t *lifespan);
34 static PyObject *nxt_py_asgi_lifespan_receive(PyObject *self, PyObject *none);
35 static PyObject *nxt_py_asgi_lifespan_send(PyObject *self, PyObject *dict);
36 static PyObject *nxt_py_asgi_lifespan_send_startup(
37     nxt_py_asgi_lifespan_t *lifespan, int v, PyObject *dict);
38 static PyObject *nxt_py_asgi_lifespan_send_(nxt_py_asgi_lifespan_t *lifespan,
39     int v, int *sent, PyObject **future);
40 static PyObject *nxt_py_asgi_lifespan_send_shutdown(
41     nxt_py_asgi_lifespan_t *lifespan, int v, PyObject *dict);
42 static PyObject *nxt_py_asgi_lifespan_disable(nxt_py_asgi_lifespan_t *lifespan);
43 static PyObject *nxt_py_asgi_lifespan_done(PyObject *self, PyObject *future);
44 
45 
46 static PyMethodDef nxt_py_asgi_lifespan_methods[] = {
47     { "receive",   nxt_py_asgi_lifespan_receive, METH_NOARGS, 0 },
48     { "send",      nxt_py_asgi_lifespan_send,    METH_O,      0 },
49     { "_done",     nxt_py_asgi_lifespan_done,    METH_O,      0 },
50     { NULL, NULL, 0, 0 }
51 };
52 
53 static PyAsyncMethods nxt_py_asgi_async_methods = {
54     .am_await = nxt_py_asgi_await,
55 };
56 
57 static PyTypeObject nxt_py_asgi_lifespan_type = {
58     PyVarObject_HEAD_INIT(NULL, 0)
59 
60     .tp_name      = "unit._asgi_lifespan",
61     .tp_basicsize = sizeof(nxt_py_asgi_lifespan_t),
62     .tp_dealloc   = nxt_py_asgi_dealloc,
63     .tp_as_async  = &nxt_py_asgi_async_methods,
64     .tp_flags     = Py_TPFLAGS_DEFAULT,
65     .tp_doc       = "unit ASGI Lifespan object",
66     .tp_iter      = nxt_py_asgi_iter,
67     .tp_iternext  = nxt_py_asgi_next,
68     .tp_methods   = nxt_py_asgi_lifespan_methods,
69 };
70 
71 
72 int
nxt_py_asgi_lifespan_startup(nxt_py_asgi_ctx_data_t * ctx_data)73 nxt_py_asgi_lifespan_startup(nxt_py_asgi_ctx_data_t *ctx_data)
74 {
75     size_t               size;
76     PyObject             *lifespan;
77     PyObject             **target_lifespans;
78     nxt_int_t            i;
79     nxt_python_target_t  *target;
80 
81     size = nxt_py_targets->count * sizeof(PyObject*);
82 
83     target_lifespans = nxt_unit_malloc(NULL, size);
84     if (nxt_slow_path(target_lifespans == NULL)) {
85         nxt_unit_alert(NULL, "Failed to allocate lifespan data");
86         return NXT_UNIT_ERROR;
87     }
88 
89     memset(target_lifespans, 0, size);
90 
91     for (i = 0; i < nxt_py_targets->count; i++) {
92         target = &nxt_py_targets->target[i];
93 
94         lifespan = nxt_py_asgi_lifespan_target_startup(ctx_data, target);
95         if (nxt_slow_path(lifespan == NULL)) {
96             return NXT_UNIT_ERROR;
97         }
98 
99         target_lifespans[i] = lifespan;
100     }
101 
102     ctx_data->target_lifespans = target_lifespans;
103 
104     return NXT_UNIT_OK;
105 }
106 
107 
108 static PyObject *
nxt_py_asgi_lifespan_target_startup(nxt_py_asgi_ctx_data_t * ctx_data,nxt_python_target_t * target)109 nxt_py_asgi_lifespan_target_startup(nxt_py_asgi_ctx_data_t *ctx_data,
110     nxt_python_target_t *target)
111 {
112     PyObject                *scope, *res, *py_task, *receive, *send, *done;
113     PyObject                *stage2;
114     nxt_py_asgi_lifespan_t  *lifespan, *ret;
115 
116     if (nxt_slow_path(PyType_Ready(&nxt_py_asgi_lifespan_type) != 0)) {
117         nxt_unit_alert(NULL,
118                  "Python failed to initialize the 'asgi_lifespan' type object");
119         return NULL;
120     }
121 
122     lifespan = PyObject_New(nxt_py_asgi_lifespan_t, &nxt_py_asgi_lifespan_type);
123     if (nxt_slow_path(lifespan == NULL)) {
124         nxt_unit_alert(NULL, "Python failed to create lifespan object");
125         return NULL;
126     }
127 
128     ret = NULL;
129 
130     receive = PyObject_GetAttrString((PyObject *) lifespan, "receive");
131     if (nxt_slow_path(receive == NULL)) {
132         nxt_unit_alert(NULL, "Python failed to get 'receive' method");
133         goto release_lifespan;
134     }
135 
136     send = PyObject_GetAttrString((PyObject *) lifespan, "send");
137     if (nxt_slow_path(receive == NULL)) {
138         nxt_unit_alert(NULL, "Python failed to get 'send' method");
139         goto release_receive;
140     }
141 
142     done = PyObject_GetAttrString((PyObject *) lifespan, "_done");
143     if (nxt_slow_path(receive == NULL)) {
144         nxt_unit_alert(NULL, "Python failed to get '_done' method");
145         goto release_send;
146     }
147 
148     lifespan->startup_future = PyObject_CallObject(ctx_data->loop_create_future,
149                                                    NULL);
150     if (nxt_slow_path(lifespan->startup_future == NULL)) {
151         nxt_unit_alert(NULL, "Python failed to create Future object");
152         nxt_python_print_exception();
153 
154         goto release_done;
155     }
156 
157     lifespan->ctx_data = ctx_data;
158     lifespan->disabled = 0;
159     lifespan->startup_received = 0;
160     lifespan->startup_sent = 0;
161     lifespan->shutdown_received = 0;
162     lifespan->shutdown_sent = 0;
163     lifespan->shutdown_called = 0;
164     lifespan->shutdown_future = NULL;
165     lifespan->receive_future = NULL;
166 
167     scope = nxt_py_asgi_new_scope(NULL, nxt_py_lifespan_str, nxt_py_2_0_str);
168     if (nxt_slow_path(scope == NULL)) {
169         goto release_future;
170     }
171 
172     if (!target->asgi_legacy) {
173         nxt_unit_req_debug(NULL, "Python call ASGI 3.0 application");
174 
175         res = PyObject_CallFunctionObjArgs(target->application,
176                                            scope, receive, send, NULL);
177 
178     } else {
179         nxt_unit_req_debug(NULL, "Python call legacy application");
180 
181         res = PyObject_CallFunctionObjArgs(target->application, scope, NULL);
182         if (nxt_slow_path(res == NULL)) {
183             nxt_unit_log(NULL, NXT_UNIT_LOG_INFO,
184                          "ASGI Lifespan processing exception");
185             nxt_python_print_exception();
186 
187             lifespan->disabled = 1;
188 
189             Py_INCREF(lifespan);
190             ret = lifespan;
191 
192             goto release_scope;
193         }
194 
195         if (nxt_slow_path(PyCallable_Check(res) == 0)) {
196             nxt_unit_req_error(NULL,
197                               "Legacy ASGI application returns not a callable");
198 
199             Py_DECREF(res);
200 
201             goto release_scope;
202         }
203 
204         stage2 = res;
205 
206         res = PyObject_CallFunctionObjArgs(stage2, receive, send, NULL);
207 
208         Py_DECREF(stage2);
209     }
210 
211     if (nxt_slow_path(res == NULL)) {
212         nxt_unit_error(NULL, "Python failed to call the application");
213         nxt_python_print_exception();
214         goto release_scope;
215     }
216 
217     if (nxt_slow_path(!PyCoro_CheckExact(res))) {
218         nxt_unit_error(NULL, "Application result type is not a coroutine");
219         Py_DECREF(res);
220         goto release_scope;
221     }
222 
223     py_task = PyObject_CallFunctionObjArgs(ctx_data->loop_create_task, res,
224                                            NULL);
225     if (nxt_slow_path(py_task == NULL)) {
226         nxt_unit_alert(NULL, "Python failed to call the create_task");
227         nxt_python_print_exception();
228         Py_DECREF(res);
229         goto release_scope;
230     }
231 
232     Py_DECREF(res);
233 
234     res = PyObject_CallMethodObjArgs(py_task, nxt_py_add_done_callback_str,
235                                      done, NULL);
236     if (nxt_slow_path(res == NULL)) {
237         nxt_unit_alert(NULL, "Python failed to call 'task.add_done_callback'");
238         nxt_python_print_exception();
239         goto release_task;
240     }
241 
242     Py_DECREF(res);
243 
244     res = PyObject_CallFunctionObjArgs(ctx_data->loop_run_until_complete,
245                                        lifespan->startup_future, NULL);
246     if (nxt_slow_path(res == NULL)) {
247         nxt_unit_alert(NULL, "Python failed to call loop.run_until_complete");
248         nxt_python_print_exception();
249         goto release_task;
250     }
251 
252     Py_DECREF(res);
253 
254     if (lifespan->startup_sent == 1 || lifespan->disabled) {
255         Py_INCREF(lifespan);
256 
257         ret = lifespan;
258     }
259 
260 release_task:
261     Py_DECREF(py_task);
262 release_scope:
263     Py_DECREF(scope);
264 release_future:
265     Py_CLEAR(lifespan->startup_future);
266 release_done:
267     Py_DECREF(done);
268 release_send:
269     Py_DECREF(send);
270 release_receive:
271     Py_DECREF(receive);
272 release_lifespan:
273     Py_DECREF(lifespan);
274 
275     return (PyObject *) ret;
276 }
277 
278 
279 int
nxt_py_asgi_lifespan_shutdown(nxt_unit_ctx_t * ctx)280 nxt_py_asgi_lifespan_shutdown(nxt_unit_ctx_t *ctx)
281 {
282     nxt_int_t               i, ret;
283     nxt_py_asgi_lifespan_t  *lifespan;
284     nxt_py_asgi_ctx_data_t  *ctx_data;
285 
286     ctx_data = ctx->data;
287 
288     for (i = 0; i < nxt_py_targets->count; i++) {
289         lifespan = (nxt_py_asgi_lifespan_t *)ctx_data->target_lifespans[i];
290 
291         ret = nxt_py_asgi_lifespan_target_shutdown(lifespan);
292         if (nxt_slow_path(ret != NXT_UNIT_OK)) {
293             return NXT_UNIT_ERROR;
294         }
295     }
296 
297     nxt_unit_free(NULL, ctx_data->target_lifespans);
298 
299     return NXT_UNIT_OK;
300 }
301 
302 
303 static int
nxt_py_asgi_lifespan_target_shutdown(nxt_py_asgi_lifespan_t * lifespan)304 nxt_py_asgi_lifespan_target_shutdown(nxt_py_asgi_lifespan_t *lifespan)
305 {
306     PyObject                *msg, *future, *res;
307     nxt_py_asgi_ctx_data_t  *ctx_data;
308 
309     ctx_data = lifespan->ctx_data;
310 
311     if (nxt_slow_path(lifespan == NULL || lifespan->disabled)) {
312         return NXT_UNIT_OK;
313     }
314 
315     lifespan->shutdown_called = 1;
316 
317     if (lifespan->receive_future != NULL) {
318         future = lifespan->receive_future;
319         lifespan->receive_future = NULL;
320 
321         msg = nxt_py_asgi_new_msg(NULL, nxt_py_lifespan_shutdown_str);
322 
323         if (nxt_fast_path(msg != NULL)) {
324             res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str,
325                                              msg, NULL);
326             Py_XDECREF(res);
327             Py_DECREF(msg);
328         }
329 
330         Py_DECREF(future);
331     }
332 
333     if (lifespan->shutdown_sent) {
334         return NXT_UNIT_OK;
335     }
336 
337     lifespan->shutdown_future = PyObject_CallObject(ctx_data->loop_create_future,
338                                                     NULL);
339     if (nxt_slow_path(lifespan->shutdown_future == NULL)) {
340         nxt_unit_alert(NULL, "Python failed to create Future object");
341         nxt_python_print_exception();
342         return NXT_UNIT_ERROR;
343     }
344 
345     res = PyObject_CallFunctionObjArgs(ctx_data->loop_run_until_complete,
346                                        lifespan->shutdown_future, NULL);
347     if (nxt_slow_path(res == NULL)) {
348         nxt_unit_alert(NULL, "Python failed to call loop.run_until_complete");
349         nxt_python_print_exception();
350         return NXT_UNIT_ERROR;
351     }
352 
353     Py_DECREF(res);
354     Py_CLEAR(lifespan->shutdown_future);
355 
356     return NXT_UNIT_OK;
357 }
358 
359 
360 static PyObject *
nxt_py_asgi_lifespan_receive(PyObject * self,PyObject * none)361 nxt_py_asgi_lifespan_receive(PyObject *self, PyObject *none)
362 {
363     PyObject                *msg, *future;
364     nxt_py_asgi_lifespan_t  *lifespan;
365     nxt_py_asgi_ctx_data_t  *ctx_data;
366 
367     lifespan = (nxt_py_asgi_lifespan_t *) self;
368     ctx_data = lifespan->ctx_data;
369 
370     nxt_unit_debug(NULL, "asgi_lifespan_receive");
371 
372     future = PyObject_CallObject(ctx_data->loop_create_future, NULL);
373     if (nxt_slow_path(future == NULL)) {
374         nxt_unit_alert(NULL, "Python failed to create Future object");
375         nxt_python_print_exception();
376 
377         return PyErr_Format(PyExc_RuntimeError,
378                             "failed to create Future object");
379     }
380 
381     if (!lifespan->startup_received) {
382         lifespan->startup_received = 1;
383 
384         msg = nxt_py_asgi_new_msg(NULL, nxt_py_lifespan_startup_str);
385 
386         return nxt_py_asgi_set_result_soon(NULL, ctx_data, future, msg);
387     }
388 
389     if (lifespan->shutdown_called && !lifespan->shutdown_received) {
390         lifespan->shutdown_received = 1;
391 
392         msg = nxt_py_asgi_new_msg(NULL, nxt_py_lifespan_shutdown_str);
393 
394         return nxt_py_asgi_set_result_soon(NULL, ctx_data, future, msg);
395     }
396 
397     Py_INCREF(future);
398     lifespan->receive_future = future;
399 
400     return future;
401 }
402 
403 
404 static PyObject *
nxt_py_asgi_lifespan_send(PyObject * self,PyObject * dict)405 nxt_py_asgi_lifespan_send(PyObject *self, PyObject *dict)
406 {
407     PyObject                *type, *msg;
408     const char              *type_str;
409     Py_ssize_t              type_len;
410     nxt_py_asgi_lifespan_t  *lifespan;
411 
412     static const nxt_str_t  startup_complete
413                                 = nxt_string("lifespan.startup.complete");
414     static const nxt_str_t  startup_failed
415                                 = nxt_string("lifespan.startup.failed");
416     static const nxt_str_t  shutdown_complete
417                                 = nxt_string("lifespan.shutdown.complete");
418     static const nxt_str_t  shutdown_failed
419                                 = nxt_string("lifespan.shutdown.failed");
420 
421     lifespan = (nxt_py_asgi_lifespan_t *) self;
422 
423     type = PyDict_GetItem(dict, nxt_py_type_str);
424     if (nxt_slow_path(type == NULL || !PyUnicode_Check(type))) {
425         nxt_unit_error(NULL,
426                        "asgi_lifespan_send: 'type' is not a unicode string");
427         return PyErr_Format(PyExc_TypeError,
428                             "'type' is not a unicode string");
429     }
430 
431     type_str = PyUnicode_AsUTF8AndSize(type, &type_len);
432 
433     nxt_unit_debug(NULL, "asgi_lifespan_send type is '%.*s'",
434                    (int) type_len, type_str);
435 
436     if (type_len == (Py_ssize_t) startup_complete.length
437         && memcmp(type_str, startup_complete.start, type_len) == 0)
438     {
439         return nxt_py_asgi_lifespan_send_startup(lifespan, 0, NULL);
440     }
441 
442     if (type_len == (Py_ssize_t) startup_failed.length
443         && memcmp(type_str, startup_failed.start, type_len) == 0)
444     {
445         msg = PyDict_GetItem(dict, nxt_py_message_str);
446         return nxt_py_asgi_lifespan_send_startup(lifespan, 1, msg);
447     }
448 
449     if (type_len == (Py_ssize_t) shutdown_complete.length
450         && memcmp(type_str, shutdown_complete.start, type_len) == 0)
451     {
452         return nxt_py_asgi_lifespan_send_shutdown(lifespan, 0, NULL);
453     }
454 
455     if (type_len == (Py_ssize_t) shutdown_failed.length
456         && memcmp(type_str, shutdown_failed.start, type_len) == 0)
457     {
458         msg = PyDict_GetItem(dict, nxt_py_message_str);
459         return nxt_py_asgi_lifespan_send_shutdown(lifespan, 1, msg);
460     }
461 
462     return nxt_py_asgi_lifespan_disable(lifespan);
463 }
464 
465 
466 static PyObject *
nxt_py_asgi_lifespan_send_startup(nxt_py_asgi_lifespan_t * lifespan,int v,PyObject * message)467 nxt_py_asgi_lifespan_send_startup(nxt_py_asgi_lifespan_t *lifespan, int v,
468     PyObject *message)
469 {
470     const char  *message_str;
471     Py_ssize_t  message_len;
472 
473     if (nxt_slow_path(v != 0)) {
474         nxt_unit_error(NULL, "Application startup failed");
475 
476         if (nxt_fast_path(message != NULL && PyUnicode_Check(message))) {
477             message_str = PyUnicode_AsUTF8AndSize(message, &message_len);
478 
479             nxt_unit_error(NULL, "%.*s", (int) message_len, message_str);
480         }
481     }
482 
483     return nxt_py_asgi_lifespan_send_(lifespan, v,
484                                       &lifespan->startup_sent,
485                                       &lifespan->startup_future);
486 }
487 
488 
489 static PyObject *
nxt_py_asgi_lifespan_send_(nxt_py_asgi_lifespan_t * lifespan,int v,int * sent,PyObject ** pfuture)490 nxt_py_asgi_lifespan_send_(nxt_py_asgi_lifespan_t *lifespan, int v, int *sent,
491     PyObject **pfuture)
492 {
493     PyObject  *future, *res;
494 
495     if (*sent) {
496         return nxt_py_asgi_lifespan_disable(lifespan);
497     }
498 
499     *sent = 1 + v;
500 
501     if (*pfuture != NULL) {
502         future = *pfuture;
503         *pfuture = NULL;
504 
505         res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str,
506                                          Py_None, NULL);
507         if (nxt_slow_path(res == NULL)) {
508             nxt_unit_alert(NULL, "Failed to call 'future.set_result'");
509             nxt_python_print_exception();
510 
511             return nxt_py_asgi_lifespan_disable(lifespan);
512         }
513 
514         Py_DECREF(res);
515         Py_DECREF(future);
516     }
517 
518     Py_INCREF(lifespan);
519 
520     return (PyObject *) lifespan;
521 }
522 
523 
524 static PyObject *
nxt_py_asgi_lifespan_disable(nxt_py_asgi_lifespan_t * lifespan)525 nxt_py_asgi_lifespan_disable(nxt_py_asgi_lifespan_t *lifespan)
526 {
527     nxt_unit_warn(NULL, "Got invalid state transition on lifespan protocol");
528 
529     lifespan->disabled = 1;
530 
531     return PyErr_Format(PyExc_AssertionError,
532                         "Got invalid state transition on lifespan protocol");
533 }
534 
535 
536 static PyObject *
nxt_py_asgi_lifespan_send_shutdown(nxt_py_asgi_lifespan_t * lifespan,int v,PyObject * message)537 nxt_py_asgi_lifespan_send_shutdown(nxt_py_asgi_lifespan_t *lifespan, int v,
538     PyObject *message)
539 {
540     return nxt_py_asgi_lifespan_send_(lifespan, v,
541                                       &lifespan->shutdown_sent,
542                                       &lifespan->shutdown_future);
543 }
544 
545 
546 static PyObject *
nxt_py_asgi_lifespan_done(PyObject * self,PyObject * future)547 nxt_py_asgi_lifespan_done(PyObject *self, PyObject *future)
548 {
549     PyObject                *res;
550     nxt_py_asgi_lifespan_t  *lifespan;
551 
552     nxt_unit_debug(NULL, "asgi_lifespan_done");
553 
554     lifespan = (nxt_py_asgi_lifespan_t *) self;
555 
556     if (lifespan->startup_sent == 0) {
557         lifespan->disabled = 1;
558     }
559 
560     /*
561      * Get Future.result() and it raises an exception, if coroutine exited
562      * with exception.
563      */
564     res = PyObject_CallMethodObjArgs(future, nxt_py_result_str, NULL);
565     if (nxt_slow_path(res == NULL)) {
566         nxt_unit_log(NULL, NXT_UNIT_LOG_INFO,
567                      "ASGI Lifespan processing exception");
568         nxt_python_print_exception();
569     }
570 
571     Py_XDECREF(res);
572 
573     if (lifespan->startup_future != NULL) {
574         future = lifespan->startup_future;
575         lifespan->startup_future = NULL;
576 
577         res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str,
578                                          Py_None, NULL);
579         if (nxt_slow_path(res == NULL)) {
580             nxt_unit_alert(NULL, "Failed to call 'future.set_result'");
581             nxt_python_print_exception();
582         }
583 
584         Py_XDECREF(res);
585         Py_DECREF(future);
586     }
587 
588     if (lifespan->shutdown_future != NULL) {
589         future = lifespan->shutdown_future;
590         lifespan->shutdown_future = NULL;
591 
592         res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str,
593                                          Py_None, NULL);
594         if (nxt_slow_path(res == NULL)) {
595             nxt_unit_alert(NULL, "Failed to call 'future.set_result'");
596             nxt_python_print_exception();
597         }
598 
599         Py_XDECREF(res);
600         Py_DECREF(future);
601     }
602 
603     Py_RETURN_NONE;
604 }
605 
606 
607 #endif /* NXT_HAVE_ASGI */
608