diff --git a/plugins/mongrel2/mongrel2.c b/plugins/mongrel2/mongrel2.c index 708a758f..c41a3c9d 100644 --- a/plugins/mongrel2/mongrel2.c +++ b/plugins/mongrel2/mongrel2.c @@ -299,6 +299,13 @@ void uwsgi_proto_zeromq_thread_fixup(struct uwsgi_socket *uwsgi_sock, int async_ // fake function, the body is in a file or completely in memory ssize_t uwsgi_proto_zeromq_read_body(struct wsgi_request *wsgi_req, char *buf, size_t len) { + size_t remains = wsgi_req->post_cl - wsgi_req->proto_parser_status; + if (remains > 0) { + if (len > remains) len = remains; + memcpy(buf, wsgi_req->proto_parser_buf + wsgi_req->proto_parser_buf_size + wsgi_req->proto_parser_status, len); + wsgi_req->proto_parser_status += len; + return len; + } return 0; } @@ -410,7 +417,8 @@ int uwsgi_proto_zeromq_accept(struct wsgi_request *wsgi_req, int fd) { } // pre-build the mongrel2 response_header - wsgi_req->proto_parser_buf = uwsgi_malloc(req_uuid_len + 1 + 11 + 1 + req_id_len + 1 + 1); + wsgi_req->proto_parser_buf_size = req_uuid_len + 1 + 11 + 1 + req_id_len + 1 + 1; + wsgi_req->proto_parser_buf = uwsgi_malloc(wsgi_req->proto_parser_buf_size); memcpy(wsgi_req->proto_parser_buf, req_uuid, req_uuid_len); ((char *) wsgi_req->proto_parser_buf)[req_uuid_len] = ' '; resp_id_len = uwsgi_num2str2(req_id_len, wsgi_req->proto_parser_buf + req_uuid_len + 1); @@ -422,13 +430,22 @@ int uwsgi_proto_zeromq_accept(struct wsgi_request *wsgi_req, int fd) { wsgi_req->proto_parser_pos = (uint64_t) req_uuid_len + 1 + resp_id_len + 1 + req_id_len + 1 + 1; // handle post data (in memory) + // reallocate wsgi_req->proto_parser_buf and change its size to be able to store request body + // the parser status holds the current position for read_body hook if (wsgi_req->post_cl > 0 && !wsgi_req->post_file) { if (uwsgi_netstring(post_data, message_size - (post_data - message_ptr), &message_ptr, &wsgi_req->post_cl)) { + char *tmp = realloc(wsgi_req->proto_parser_buf, wsgi_req->proto_parser_buf_size + wsgi_req->post_cl); + if (!tmp) { + uwsgi_error("realloc()"); + exit(1); + } + wsgi_req->proto_parser_buf = tmp; + // status is an offset... + wsgi_req->proto_parser_status = 0; #ifdef UWSGI_DEBUG uwsgi_log("post_size: %d\n", wsgi_req->post_cl); #endif - wsgi_req->post_read_buf = uwsgi_malloc(wsgi_req->post_cl); - memcpy(wsgi_req->post_read_buf, message_ptr, wsgi_req->post_cl); + memcpy(wsgi_req->proto_parser_buf + wsgi_req->proto_parser_buf_size, message_ptr, wsgi_req->post_cl); } } @@ -452,33 +469,7 @@ retry: return -1; } -void uwsgi_proto_zeromq_close(struct wsgi_request *wsgi_req) { - zmq_msg_t reply; - - // check for already freed wsgi_req->proto_parser_buf/wsgi_req->proto_parser_pos - if (!wsgi_req->proto_parser_pos) - return; - - // no need to pass a free function (the buffer will be freed during cloe_request) - zmq_msg_init_data(&reply, wsgi_req->proto_parser_buf, wsgi_req->proto_parser_pos, NULL, NULL); - if (uwsgi.threads > 1) - pthread_mutex_lock(&wsgi_req->socket->lock); -#if ZMQ_VERSION >= ZMQ_MAKE_VERSION(3,0,0) - if (zmq_sendmsg(wsgi_req->socket->pub, &reply, 0)) { - uwsgi_error("uwsgi_proto_zeromq_close()/zmq_sendmsg()"); -#else - if (zmq_send(wsgi_req->socket->pub, &reply, 0)) { - uwsgi_error("uwsgi_proto_zeromq_close()/zmq_send()"); -#endif - } - if (uwsgi.threads > 1) - pthread_mutex_unlock(&wsgi_req->socket->lock); - zmq_msg_close(&reply); - -} - - -int uwsgi_proto_zeromq_write(struct wsgi_request *wsgi_req, char *buf, size_t len) { +static int uwsgi_proto_zeromq_write_do(struct wsgi_request *wsgi_req, char *buf, size_t len) { zmq_msg_t reply; if (zmq_msg_init_size(&reply, wsgi_req->proto_parser_pos + len)) { @@ -489,7 +480,8 @@ int uwsgi_proto_zeromq_write(struct wsgi_request *wsgi_req, char *buf, size_t le char *zmq_body = zmq_msg_data(&reply); memcpy(zmq_body, wsgi_req->proto_parser_buf, wsgi_req->proto_parser_pos); - memcpy(zmq_body + wsgi_req->proto_parser_pos, buf, len); + if (len > 0) + memcpy(zmq_body + wsgi_req->proto_parser_pos, buf, len); if (uwsgi.threads > 1) pthread_mutex_lock(&wsgi_req->socket->lock); @@ -510,6 +502,18 @@ int uwsgi_proto_zeromq_write(struct wsgi_request *wsgi_req, char *buf, size_t le return UWSGI_OK; } +int uwsgi_proto_zeromq_write(struct wsgi_request *wsgi_req, char *buf, size_t len) { + int ret = uwsgi_proto_zeromq_write_do(wsgi_req, buf, len); + if (ret == UWSGI_OK) { + wsgi_req->write_pos += len; + } + return ret; +} + +void uwsgi_proto_zeromq_close(struct wsgi_request *wsgi_req) { + uwsgi_proto_zeromq_write_do(wsgi_req, "", 0); +} + /* we have a problem... recent Mongrel2 releases introduced a ring buffer that limit the amount of messages we can send (or better, the amount of @@ -532,7 +536,7 @@ int uwsgi_proto_zeromq_sendfile(struct wsgi_request *wsgi_req, int fd, size_t po return -1; } wsgi_req->write_pos += rlen; - if (uwsgi_proto_zeromq_write(wsgi_req, tmp_buf, rlen) < 0) { + if (uwsgi_proto_zeromq_write_do(wsgi_req, tmp_buf, rlen) < 0) { free(tmp_buf); return -1; }