Resonse body got cut off when containing binary image data using asgi protocol

[synodriver]

Hi,
When switching to unit, I found some imcompatible
Here is an example:

from pathlib import Path

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


@app.get("/")
async def _():
    f = open(str(Path(__file__).parent / "a.png"), "rb")  
    # It's just an example, actually it's generated in code instead of read from file
    return StreamingResponse(f, media_type="image/png")

Put a picture a.png in it's path, any picture is ok for that

My environment

The command unitd --version
got

unit version: 1.24.0
configured as ./configure --ld-opt=-fno-lto  --openssl

The version of fastapi and python

Python 3.8.5 
fastapi 0.63.0
uvicorn 0.14.0
hypercorn 0.9.2   

then try other asgi servers and unit to start a server using the same code

uvicorn main:app --port 9000
hypercorn main:app --bind 127.0.0.1:9001

the unit config

{
  "listeners": {
    "*:9002": {
      "pass": "applications/fastapi"
    }
  },
  "applications": {
    "fastapi": {
      "type": "python 3.8",
      "path": "path to app",
      "module": "main",
      "callable": "app"
    }
  }
}

Open http://127.0.0.1:9000 http://127.0.0.1:9001 http://127.0.0.1:9002 in browser and that's the difference

I did some research about this, found that unit may be mistake '\n' as the end of response.

I tested them using the folowing code

import requests

uvicorn_resp = requests.get("http://127.0.0.1:9000")
hypercorn_resp = requests.get("http://127.0.0.1:9001")
unit_resp = requests.get("http://127.0.0.1:9002")

print(f"uvicorn_resp length:{len(uvicorn_resp.content)}")
print(f"hypercorn_resp length:{len(hypercorn_resp.content)}")
print(f"unit_resp length:{len(unit_resp.content)}")

print(unit_resp.content)
assert chr(unit_resp.content[-1]) == "\n"

I got

uvicorn_resp length:142447
hypercorn_resp length:142447
unit_resp length:26
b'\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00\xff\xdb\x00C\x00\n'

It is obvious that response body of unit was cut off, while the other two asgi server doesn't.

Any Suggestion about that?

[mar0x]

Hello.
Thank you for your report!

There are bugs in Unit ASGI implementation:

• receive() for GET request never blocks;
• receive() returns http.disconnect after response headers sent.

Please try the following patch:

diff --git a/src/python/nxt_python_asgi_http.c b/src/python/nxt_python_asgi_http.c
--- a/src/python/nxt_python_asgi_http.c
+++ b/src/python/nxt_python_asgi_http.c
@@ -27,6 +27,7 @@ typedef struct {
     int                      closed;
     PyObject                 *send_body;
     Py_ssize_t               send_body_off;
+    int                      empty_body_received;
 } nxt_py_asgi_http_t;
 
 
@@ -37,6 +38,7 @@ static PyObject *nxt_py_asgi_http_respon
     PyObject *dict);
 static PyObject *nxt_py_asgi_http_response_body(nxt_py_asgi_http_t *http,
     PyObject *dict);
+static void nxt_py_asgi_http_emit_disconnect(nxt_py_asgi_http_t *http);
 static PyObject *nxt_py_asgi_http_done(PyObject *self, PyObject *future);
 
 
@@ -98,6 +100,7 @@ nxt_py_asgi_http_create(nxt_unit_request
         http->closed = 0;
         http->send_body = NULL;
         http->send_body_off = 0;
+        http->empty_body_received = 0;
     }
 
     return (PyObject *) http;
@@ -117,7 +120,7 @@ nxt_py_asgi_http_receive(PyObject *self,
 
     nxt_unit_req_debug(req, "asgi_http_receive");
 
-    if (nxt_slow_path(http->closed || nxt_unit_response_is_sent(req))) {
+    if (nxt_slow_path(http->closed || http->complete )) {
         msg = nxt_py_asgi_new_msg(req, nxt_py_http_disconnect_str);
 
     } else {
@@ -171,6 +174,14 @@ nxt_py_asgi_http_read_msg(nxt_py_asgi_ht
         size = nxt_py_asgi_http_body_buf_size;
     }
 
+    if (size == 0) {
+        if (http->empty_body_received) {
+            Py_RETURN_NONE;
+        }
+
+        http->empty_body_received = 1;
+    }
+
     if (size > 0) {
         body = PyBytes_FromStringAndSize(NULL, size);
         if (nxt_slow_path(body == NULL)) {
@@ -442,6 +453,8 @@ nxt_py_asgi_http_response_body(nxt_py_as
 
     if (more_body == NULL || more_body == Py_False) {
         http->complete = 1;
+
+        nxt_py_asgi_http_emit_disconnect(http);
     }
 
     Py_INCREF(http);
@@ -449,6 +462,41 @@ nxt_py_asgi_http_response_body(nxt_py_as
 }
 
 
+static void
+nxt_py_asgi_http_emit_disconnect(nxt_py_asgi_http_t *http)
+{
+    PyObject  *msg, *future, *res;
+
+    if (http->receive_future == NULL) {
+        return;
+    }
+
+    msg = nxt_py_asgi_new_msg(http->req, nxt_py_http_disconnect_str);
+    if (nxt_slow_path(msg == NULL)) {
+        return;
+    }
+
+    if (msg == Py_None) {
+        Py_DECREF(msg);
+        return;
+    }
+
+    future = http->receive_future;
+    http->receive_future = NULL;
+
+    res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg, NULL);
+    if (nxt_slow_path(res == NULL)) {
+        nxt_unit_req_alert(http->req, "'set_result' call failed");
+        nxt_python_print_exception();
+    }
+
+    Py_XDECREF(res);
+    Py_DECREF(future);
+
+    Py_DECREF(msg);
+}
+
+
 void
 nxt_py_asgi_http_data_handler(nxt_unit_request_info_t *req)
 {
@@ -573,7 +621,6 @@ fail:
 void
 nxt_py_asgi_http_close_handler(nxt_unit_request_info_t *req)
 {
-    PyObject            *msg, *future, *res;
     nxt_py_asgi_http_t  *http;
 
     http = req->data;
@@ -582,33 +629,7 @@ nxt_py_asgi_http_close_handler(nxt_unit_
 
     http->closed = 1;
 
-    if (http->receive_future == NULL) {
-        return;
-    }
-
-    msg = nxt_py_asgi_new_msg(req, nxt_py_http_disconnect_str);
-    if (nxt_slow_path(msg == NULL)) {
-        return;
-    }
-
-    if (msg == Py_None) {
-        Py_DECREF(msg);
-        return;
-    }
-
-    future = http->receive_future;
-    http->receive_future = NULL;
-
-    res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg, NULL);
-    if (nxt_slow_path(res == NULL)) {
-        nxt_unit_req_alert(req, "'set_result' call failed");
-        nxt_python_print_exception();
-    }
-
-    Py_XDECREF(res);
-    Py_DECREF(future);
-
-    Py_DECREF(msg);
+    nxt_py_asgi_http_emit_disconnect(http);
 }
 
 

[synodriver]

Hi,
The patch works!
But sometimes the log will get

Exception in callback Future.set_result({'type': 'http.disconnect'})
handle: <Handle Future.set_result({'type': 'http.disconnect'})>
Traceback (most recent call last):
  File "/root/miniconda3/lib/python3.8/asyncio/events.py", line 81, in _run
    self._context.run(self._callback, *self._args)
asyncio.exceptions.InvalidStateError: invalid state

And this also happens before applying the patch.
Since it didn't influence the website and quite hard to reproduce, is it ok to ignore this exception?

[mar0x]

Thanks for the update.
Could you please check if there is a string 'set_result' call failed in a log file close to the exception information?

[synodriver]

Thanks for the update.
Could you please check if there is a string 'set_result' call failed in a log file close to the exception information?

Yes, I found three 'set_result' call failed in the log file while there are more InvalidStateError

[mar0x]

Ok. Please apply the patch below (over the first patch):

diff --git a/src/python/nxt_python_asgi_http.c b/src/python/nxt_python_asgi_http.c
--- a/src/python/nxt_python_asgi_http.c
+++ b/src/python/nxt_python_asgi_http.c
@@ -39,6 +39,8 @@ static PyObject *nxt_py_asgi_http_respon
 static PyObject *nxt_py_asgi_http_response_body(nxt_py_asgi_http_t *http,
     PyObject *dict);
 static void nxt_py_asgi_http_emit_disconnect(nxt_py_asgi_http_t *http);
+static void nxt_py_asgi_http_set_result(nxt_py_asgi_http_t *http,
+    PyObject *future, PyObject *msg);
 static PyObject *nxt_py_asgi_http_done(PyObject *self, PyObject *future);
 
 
@@ -465,7 +467,7 @@ nxt_py_asgi_http_response_body(nxt_py_as
 static void
 nxt_py_asgi_http_emit_disconnect(nxt_py_asgi_http_t *http)
 {
-    PyObject  *msg, *future, *res;
+    PyObject  *msg, *future;
 
     if (http->receive_future == NULL) {
         return;
@@ -484,23 +486,45 @@ nxt_py_asgi_http_emit_disconnect(nxt_py_
     future = http->receive_future;
     http->receive_future = NULL;
 
-    res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg, NULL);
+    nxt_py_asgi_http_set_result(http, future, msg);
+
+    Py_DECREF(msg);
+}
+
+
+static void
+nxt_py_asgi_http_set_result(nxt_py_asgi_http_t *http, PyObject *future,
+    PyObject *msg)
+{
+    PyObject  *res;
+
+    res = PyObject_CallMethodObjArgs(future, nxt_py_done_str, NULL);
     if (nxt_slow_path(res == NULL)) {
-        nxt_unit_req_alert(http->req, "'set_result' call failed");
+        nxt_unit_req_alert(http->req, "'done' call failed");
         nxt_python_print_exception();
     }
 
+    if (res == Py_False) {
+        res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg,
+                                         NULL);
+        if (nxt_slow_path(res == NULL)) {
+            nxt_unit_req_alert(http->req, "'set_result' call failed");
+            nxt_python_print_exception();
+        }
+
+    } else {
+        res = NULL;
+    }
+
     Py_XDECREF(res);
     Py_DECREF(future);
-
-    Py_DECREF(msg);
 }
 
 
 void
 nxt_py_asgi_http_data_handler(nxt_unit_request_info_t *req)
 {
-    PyObject            *msg, *future, *res;
+    PyObject            *msg, *future;
     nxt_py_asgi_http_t  *http;
 
     http = req->data;
@@ -524,14 +548,7 @@ nxt_py_asgi_http_data_handler(nxt_unit_r
     future = http->receive_future;
     http->receive_future = NULL;
 
-    res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, msg, NULL);
-    if (nxt_slow_path(res == NULL)) {
-        nxt_unit_req_alert(req, "'set_result' call failed");
-        nxt_python_print_exception();
-    }
-
-    Py_XDECREF(res);
-    Py_DECREF(future);
+    nxt_py_asgi_http_set_result(http, future, msg);
 
     Py_DECREF(msg);
 }
@@ -575,15 +592,7 @@ nxt_py_asgi_http_drain(nxt_queue_link_t 
     future = http->send_future;
     http->send_future = NULL;
 
-    res = PyObject_CallMethodObjArgs(future, nxt_py_set_result_str, Py_None,
-                                     NULL);
-    if (nxt_slow_path(res == NULL)) {
-        nxt_unit_req_alert(http->req, "'set_result' call failed");
-        nxt_python_print_exception();
-    }
-
-    Py_XDECREF(res);
-    Py_DECREF(future);
+    nxt_py_asgi_http_set_result(http, future, Py_None);
 
     return NXT_UNIT_OK;
 

This should fix the exceptions.
Starlette cancels the future which returned by receive() call and with this patch Unit will check the done state of future before call set_result().

[synodriver]

The same exception keeps occur after apply the patch, however, the websites just works fine. And I didn't see any 'set_result' call failed or 'done' call failed in the log file this time.

[mar0x]

I'm ready to reproduce and try to fix it if you provide an application sample (or mockup).

[synodriver]

I'm ready to reproduce and try to fix it if you provide an application sample (or mockup).

Actually it's an api receives uplaod pictures and appraisal it using TensorFlow, however, the asyncio.exceptions.InvalidStateError only happens when used with tensorflow. How strange it is! I've tried to reproduce it by replacing tensorflow code to time.sleep but failed. Maybe we can just ignore it in this case?

The code looks like

from fastapi import File, UploadFile, Form, FastAPI

app=FastAPI()

@app.post("/nsfw")
def process_pic(data: UploadFile = File(...),
                image_dim: int = Form(224)):
    ret = classify(data.file, image_dim)  # the tenserflow code
    return ret

Sorry about that. It seems that that exception comes from my code instead of unit. although it works fine on uvicorn
Maybe the multithreading in python caused it?

[synodriver]

I will also make that strange code public later. Be aware of it because it relys on many scientific packages, which may be hard to install.

[synodriver]

Thanks for your great help on this issue (may be two?).
After hours of trying to reproduce the exception, I realized how hard it is to reproduce a bug from a complicated application. Anyway I will ignore the exception now since it has no effect except making log file grow faster. Maybe it's caused by threading since asyncio is not thread safe, but for me, this two patch is stable enough for production now.
So will this two patch merged to master branch? I would be happy to see them in the next release.

[synodriver]

Now I can confirm that is tensorflow caused it (: because without tf the exception won't happen
Full log

2021/07/09 13:45:31 [notice] 32232#32232 process 27019 exited with code 0
2021/07/09 13:45:39 [info] 27256#27256 "fastapi" application started
Exception in callback Future.set_result({'type': 'http.disconnect'})
handle: <Handle Future.set_result({'type': 'http.disconnect'})>
Traceback (most recent call last):
  File "/root/miniconda3/lib/python3.8/asyncio/events.py", line 81, in _run
    self._context.run(self._callback, *self._args)
asyncio.exceptions.InvalidStateError: invalid state
2021-07-09 13:45:41.354407: I tensorflow/core/platform/cpu_feature_guard.cc:142] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN)to use the following CPU instructions in performance-critical operations:  SSE4.1 SSE4.2 AVX AVX2 FMA
To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags.
2021-07-09 13:45:41.359472: I tensorflow/core/platform/profile_utils/cpu_utils.cc:104] CPU Frequency: 2595120000 Hz
2021-07-09 13:45:41.359730: I tensorflow/compiler/xla/service/service.cc:168] XLA service 0x558a50d3b520 initialized for platform Host (this does not guarantee that XLA will be used). Devices:
2021-07-09 13:45:41.359753: I tensorflow/compiler/xla/service/service.cc:176]   StreamExecutor device (0): Host, Default Version
WARNING:tensorflow:Error in loading the saved optimizer state. As a result, your model is starting with a freshly initialized optimizer.
Exception in callback Future.set_result({'type': 'http.disconnect'})
handle: <Handle Future.set_result({'type': 'http.disconnect'})>
Traceback (most recent call last):
  File "/root/miniconda3/lib/python3.8/asyncio/events.py", line 81, in _run
    self._context.run(self._callback, *self._args)
asyncio.exceptions.InvalidStateError: invalid state
Exception in callback Future.set_result({'type': 'http.disconnect'})
handle: <Handle Future.set_result({'type': 'http.disconnect'})>
Traceback (most recent call last):
  File "/root/miniconda3/lib/python3.8/asyncio/events.py", line 81, in _run
    self._context.run(self._callback, *self._args)
asyncio.exceptions.InvalidStateError: invalid state
2021/07/09 13:46:03 [notice] 32232#32232 process 26973 exited with code 0

For those who don't deploy deeplearning project, it's a good news.

[mar0x]

So, (it still not clear for me) is the exception present, when you run the application in uvicorn?

If "yes", then (most likely) it is impossible to fix it in Unit :)
If "no", then there is another known difference (rather a feature) in Unit ASGI implementation: it creates new event loop to run application for each thread, even for the main one. This feature may cause unexpected exceptions or strange bugs when asyncio synchronisation primitives initialised before application starts (e.g. globally).

Although I like the approach with new event loop for main thread because it is consistent and helps to prepare application to run in multiple threads, I admit it can be source of pain for people who just want to run single-threaded ASGI application in Unit.

In view of the amount of work you've done trying to fix the issue, please try another one patch for Unit, which should use default event loop in main thread:

diff --git a/src/python/nxt_python.c b/src/python/nxt_python.c
--- a/src/python/nxt_python.c
+++ b/src/python/nxt_python.c
@@ -264,7 +264,7 @@ nxt_python_start(nxt_task_t *task, nxt_p
         goto fail;
     }
 
-    rc = nxt_py_proto.ctx_data_alloc(&python_init.ctx_data);
+    rc = nxt_py_proto.ctx_data_alloc(&python_init.ctx_data, 0);
     if (nxt_slow_path(rc != NXT_UNIT_OK)) {
         goto fail;
     }
@@ -504,7 +504,7 @@ nxt_python_init_threads(nxt_python_app_c
     for (i = 0; i < c->threads - 1; i++) {
         ti = &nxt_py_threads[i];
 
-        res = nxt_py_proto.ctx_data_alloc(&ti->ctx_data);
+        res = nxt_py_proto.ctx_data_alloc(&ti->ctx_data, i + 1);
         if (nxt_slow_path(res != NXT_UNIT_OK)) {
             return NXT_UNIT_ERROR;
         }
diff --git a/src/python/nxt_python.h b/src/python/nxt_python.h
--- a/src/python/nxt_python.h
+++ b/src/python/nxt_python.h
@@ -60,7 +60,7 @@ typedef struct {
 
 
 typedef struct {
-    int   (*ctx_data_alloc)(void **pdata);
+    int   (*ctx_data_alloc)(void **pdata, uint32_t id);
     void  (*ctx_data_free)(void *data);
     int   (*startup)(void *data);
     int   (*run)(nxt_unit_ctx_t *ctx);
diff --git a/src/python/nxt_python_asgi.c b/src/python/nxt_python_asgi.c
--- a/src/python/nxt_python_asgi.c
+++ b/src/python/nxt_python_asgi.c
@@ -17,7 +17,7 @@
 
 
 static PyObject *nxt_python_asgi_get_func(PyObject *obj);
-static int nxt_python_asgi_ctx_data_alloc(void **pdata);
+static int nxt_python_asgi_ctx_data_alloc(void **pdata, uint32_t id);
 static void nxt_python_asgi_ctx_data_free(void *data);
 static int nxt_python_asgi_startup(void *data);
 static int nxt_python_asgi_run(nxt_unit_ctx_t *ctx);
@@ -194,10 +194,11 @@ nxt_python_asgi_init(nxt_unit_init_t *in
 
 
 static int
-nxt_python_asgi_ctx_data_alloc(void **pdata)
+nxt_python_asgi_ctx_data_alloc(void **pdata, uint32_t id)
 {
     uint32_t                i;
-    PyObject                *asyncio, *loop, *new_event_loop, *obj;
+    PyObject                *asyncio, *loop, *event_loop, *obj;
+    const char              *event_loop_func;
     nxt_py_asgi_ctx_data_t  *ctx_data;
 
     ctx_data = nxt_unit_malloc(NULL, sizeof(nxt_py_asgi_ctx_data_t));
@@ -232,23 +233,28 @@ nxt_python_asgi_ctx_data_alloc(void **pd
         goto fail;
     }
 
-    new_event_loop = PyDict_GetItemString(PyModule_GetDict(asyncio),
-                                          "new_event_loop");
-    if (nxt_slow_path(new_event_loop == NULL)) {
+    event_loop_func = id == 0 ? "get_event_loop" : "new_event_loop";
+
+    event_loop = PyDict_GetItemString(PyModule_GetDict(asyncio),
+                                      event_loop_func);
+    if (nxt_slow_path(event_loop == NULL)) {
         nxt_unit_alert(NULL,
-                 "Python failed to get 'new_event_loop' from module 'asyncio'");
+                       "Python failed to get '%s' from module 'asyncio'",
+                       event_loop_func);
         goto fail;
     }
 
-    if (nxt_slow_path(PyCallable_Check(new_event_loop) == 0)) {
+    if (nxt_slow_path(PyCallable_Check(event_loop) == 0)) {
         nxt_unit_alert(NULL,
-                       "'asyncio.new_event_loop' is not a callable object");
+                       "'asyncio.%s' is not a callable object",
+                       event_loop_func);
         goto fail;
     }
 
-    loop = PyObject_CallObject(new_event_loop, NULL);
+    loop = PyObject_CallObject(event_loop, NULL);
     if (nxt_slow_path(loop == NULL)) {
-        nxt_unit_alert(NULL, "Python failed to call 'asyncio.new_event_loop'");
+        nxt_unit_alert(NULL, "Python failed to call 'asyncio.%s'",
+                       event_loop_func);
         goto fail;
     }
 
diff --git a/src/python/nxt_python_wsgi.c b/src/python/nxt_python_wsgi.c
--- a/src/python/nxt_python_wsgi.c
+++ b/src/python/nxt_python_wsgi.c
@@ -51,7 +51,7 @@ typedef struct {
 }  nxt_python_ctx_t;
 
 
-static int nxt_python_wsgi_ctx_data_alloc(void **pdata);
+static int nxt_python_wsgi_ctx_data_alloc(void **pdata, uint32_t id);
 static void nxt_python_wsgi_ctx_data_free(void *data);
 static int nxt_python_wsgi_run(nxt_unit_ctx_t *ctx);
 static void nxt_python_wsgi_done(void);
@@ -210,7 +210,7 @@ fail:
 
 
 static int
-nxt_python_wsgi_ctx_data_alloc(void **pdata)
+nxt_python_wsgi_ctx_data_alloc(void **pdata, uint32_t id)
 {
     nxt_python_ctx_t  *pctx;
 

I'm appreciate your effort and feedback for Unit project: 3 new issues in ASGI implementation were caught and submitted for review.
Thank you!

[synodriver]

Hi,
The application runs fine without any exception on uvicorn.
After this patch, it still raises asyncio.exceptions.InvalidStateError: invalid state
I think no one will know the bug without looking into the log file, so maybe we should ignore such condition like this? Besides, it's quite hard to reproduce.
And for any one who may concern about the strange code, here is one