diff options
Diffstat (limited to 'src/nxt_unit.c')
| -rw-r--r-- | src/nxt_unit.c | 1159 |
1 files changed, 888 insertions, 271 deletions
diff --git a/src/nxt_unit.c b/src/nxt_unit.c index 88c7fa6a..28a0de20 100644 --- a/src/nxt_unit.c +++ b/src/nxt_unit.c @@ -11,38 +11,60 @@ #include "nxt_unit.h" #include "nxt_unit_request.h" #include "nxt_unit_response.h" +#include "nxt_unit_websocket.h" + +#include "nxt_websocket.h" #if (NXT_HAVE_MEMFD_CREATE) #include <linux/memfd.h> #endif -typedef struct nxt_unit_impl_s nxt_unit_impl_t; -typedef struct nxt_unit_mmap_s nxt_unit_mmap_t; -typedef struct nxt_unit_mmaps_s nxt_unit_mmaps_t; -typedef struct nxt_unit_process_s nxt_unit_process_t; -typedef struct nxt_unit_mmap_buf_s nxt_unit_mmap_buf_t; -typedef struct nxt_unit_recv_msg_s nxt_unit_recv_msg_t; -typedef struct nxt_unit_ctx_impl_s nxt_unit_ctx_impl_t; -typedef struct nxt_unit_port_impl_s nxt_unit_port_impl_t; -typedef struct nxt_unit_request_info_impl_s nxt_unit_request_info_impl_t; +typedef struct nxt_unit_impl_s nxt_unit_impl_t; +typedef struct nxt_unit_mmap_s nxt_unit_mmap_t; +typedef struct nxt_unit_mmaps_s nxt_unit_mmaps_t; +typedef struct nxt_unit_process_s nxt_unit_process_t; +typedef struct nxt_unit_mmap_buf_s nxt_unit_mmap_buf_t; +typedef struct nxt_unit_recv_msg_s nxt_unit_recv_msg_t; +typedef struct nxt_unit_ctx_impl_s nxt_unit_ctx_impl_t; +typedef struct nxt_unit_port_impl_s nxt_unit_port_impl_t; +typedef struct nxt_unit_request_info_impl_s nxt_unit_request_info_impl_t; +typedef struct nxt_unit_websocket_frame_impl_s nxt_unit_websocket_frame_impl_t; static nxt_unit_impl_t *nxt_unit_create(nxt_unit_init_t *init); static void nxt_unit_ctx_init(nxt_unit_impl_t *lib, nxt_unit_ctx_impl_t *ctx_impl, void *data); +nxt_inline void nxt_unit_mmap_buf_insert(nxt_unit_mmap_buf_t **head, + nxt_unit_mmap_buf_t *mmap_buf); +nxt_inline void nxt_unit_mmap_buf_insert_tail(nxt_unit_mmap_buf_t **prev, + nxt_unit_mmap_buf_t *mmap_buf); +nxt_inline void nxt_unit_mmap_buf_remove(nxt_unit_mmap_buf_t *mmap_buf); static int nxt_unit_read_env(nxt_unit_port_t *ready_port, nxt_unit_port_t *read_port, int *log_fd, uint32_t *stream); static int nxt_unit_ready(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id, uint32_t stream); +static int nxt_unit_process_new_port(nxt_unit_ctx_t *ctx, + nxt_unit_recv_msg_t *recv_msg); +static int nxt_unit_process_req_headers(nxt_unit_ctx_t *ctx, + nxt_unit_recv_msg_t *recv_msg); +static int nxt_unit_process_websocket(nxt_unit_ctx_t *ctx, + nxt_unit_recv_msg_t *recv_msg); static nxt_unit_request_info_impl_t *nxt_unit_request_info_get( nxt_unit_ctx_t *ctx); static void nxt_unit_request_info_release(nxt_unit_request_info_t *req); static void nxt_unit_request_info_free(nxt_unit_request_info_impl_t *req); +static nxt_unit_websocket_frame_impl_t *nxt_unit_websocket_frame_get( + nxt_unit_ctx_t *ctx); +static void nxt_unit_websocket_frame_release(nxt_unit_websocket_frame_t *ws); +static void nxt_unit_websocket_frame_free(nxt_unit_websocket_frame_impl_t *ws); static nxt_unit_process_t *nxt_unit_msg_get_process(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg); static nxt_unit_mmap_buf_t *nxt_unit_mmap_buf_get(nxt_unit_ctx_t *ctx); static void nxt_unit_mmap_buf_release(nxt_unit_mmap_buf_t *mmap_buf); static int nxt_unit_mmap_buf_send(nxt_unit_ctx_t *ctx, uint32_t stream, nxt_unit_mmap_buf_t *mmap_buf, int last); +static void nxt_unit_mmap_buf_free(nxt_unit_mmap_buf_t *mmap_buf); +static ssize_t nxt_unit_buf_read(nxt_unit_buf_t **b, uint64_t *len, void *dst, + size_t size); static nxt_port_mmap_header_t *nxt_unit_mmap_get(nxt_unit_ctx_t *ctx, nxt_unit_process_t *process, nxt_unit_port_id_t *port_id, nxt_chunk_id_t *c, int n); @@ -65,7 +87,7 @@ static nxt_port_mmap_header_t *nxt_unit_get_incoming_mmap(nxt_unit_ctx_t *ctx, static int nxt_unit_tracking_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg); static int nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, - nxt_unit_recv_msg_t *recv_msg, nxt_queue_t *incoming_buf); + nxt_unit_recv_msg_t *recv_msg); static int nxt_unit_mmap_release(nxt_port_mmap_header_t *hdr, void *start, uint32_t size); @@ -98,14 +120,22 @@ static int nxt_unit_port_hash_add(nxt_lvlhsh_t *port_hash, static nxt_unit_port_impl_t *nxt_unit_port_hash_find(nxt_lvlhsh_t *port_hash, nxt_unit_port_id_t *port_id, int remove); +static int nxt_unit_request_hash_add(nxt_lvlhsh_t *request_hash, + nxt_unit_request_info_impl_t *req_impl); +static nxt_unit_request_info_impl_t *nxt_unit_request_hash_find( + nxt_lvlhsh_t *request_hash, uint32_t stream, int remove); + static char * nxt_unit_snprint_prefix(char *p, char *end, pid_t pid, int level); struct nxt_unit_mmap_buf_s { nxt_unit_buf_t buf; + nxt_unit_mmap_buf_t *next; + nxt_unit_mmap_buf_t **prev; + nxt_port_mmap_header_t *hdr; - nxt_queue_link_t link; +// nxt_queue_link_t link; nxt_unit_port_id_t port_id; nxt_unit_request_info_t *req; nxt_unit_ctx_impl_t *ctx_impl; @@ -113,12 +143,20 @@ struct nxt_unit_mmap_buf_s { struct nxt_unit_recv_msg_s { - nxt_port_msg_t port_msg; + uint32_t stream; + nxt_pid_t pid; + nxt_port_id_t reply_port; + + uint8_t last; /* 1 bit */ + uint8_t mmap; /* 1 bit */ void *start; uint32_t size; + int fd; nxt_unit_process_t *process; + + nxt_unit_mmap_buf_t *incoming_buf; }; @@ -127,18 +165,22 @@ typedef enum { NXT_UNIT_RS_RESPONSE_INIT, NXT_UNIT_RS_RESPONSE_HAS_CONTENT, NXT_UNIT_RS_RESPONSE_SENT, - NXT_UNIT_RS_DONE, + NXT_UNIT_RS_RELEASED, } nxt_unit_req_state_t; struct nxt_unit_request_info_impl_s { nxt_unit_request_info_t req; - nxt_unit_recv_msg_t recv_msg; - nxt_queue_t outgoing_buf; /* of nxt_unit_mmap_buf_t */ - nxt_queue_t incoming_buf; /* of nxt_unit_mmap_buf_t */ + uint32_t stream; + + nxt_unit_process_t *process; + + nxt_unit_mmap_buf_t *outgoing_buf; + nxt_unit_mmap_buf_t *incoming_buf; nxt_unit_req_state_t state; + uint8_t websocket; nxt_queue_link_t link; @@ -146,6 +188,19 @@ struct nxt_unit_request_info_impl_s { }; +struct nxt_unit_websocket_frame_impl_s { + nxt_unit_websocket_frame_t ws; + + nxt_unit_mmap_buf_t *buf; + + nxt_queue_link_t link; + + nxt_unit_ctx_impl_t *ctx_impl; + + void *retain_buf; +}; + + struct nxt_unit_ctx_impl_s { nxt_unit_ctx_t ctx; @@ -154,14 +209,20 @@ struct nxt_unit_ctx_impl_s { nxt_queue_link_t link; - nxt_queue_t free_buf; /* of nxt_unit_mmap_buf_t */ + nxt_unit_mmap_buf_t *free_buf; /* of nxt_unit_request_info_impl_t */ nxt_queue_t free_req; + /* of nxt_unit_websocket_frame_impl_t */ + nxt_queue_t free_ws; + /* of nxt_unit_request_info_impl_t */ nxt_queue_t active_req; + /* of nxt_unit_request_info_impl_t */ + nxt_lvlhsh_t requests; + nxt_unit_mmap_buf_t ctx_buf[2]; nxt_unit_request_info_impl_t req; @@ -394,18 +455,65 @@ nxt_unit_ctx_init(nxt_unit_impl_t *lib, nxt_unit_ctx_impl_t *ctx_impl, nxt_queue_insert_tail(&lib->contexts, &ctx_impl->link); - nxt_queue_init(&ctx_impl->free_buf); nxt_queue_init(&ctx_impl->free_req); + nxt_queue_init(&ctx_impl->free_ws); nxt_queue_init(&ctx_impl->active_req); - nxt_queue_insert_tail(&ctx_impl->free_buf, &ctx_impl->ctx_buf[0].link); - nxt_queue_insert_tail(&ctx_impl->free_buf, &ctx_impl->ctx_buf[1].link); + ctx_impl->free_buf = NULL; + nxt_unit_mmap_buf_insert(&ctx_impl->free_buf, &ctx_impl->ctx_buf[1]); + nxt_unit_mmap_buf_insert(&ctx_impl->free_buf, &ctx_impl->ctx_buf[0]); + nxt_queue_insert_tail(&ctx_impl->free_req, &ctx_impl->req.link); ctx_impl->req.req.ctx = &ctx_impl->ctx; ctx_impl->req.req.unit = &lib->unit; ctx_impl->read_port_fd = -1; + ctx_impl->requests.slot = 0; +} + + +nxt_inline void +nxt_unit_mmap_buf_insert(nxt_unit_mmap_buf_t **head, + nxt_unit_mmap_buf_t *mmap_buf) +{ + mmap_buf->next = *head; + + if (mmap_buf->next != NULL) { + mmap_buf->next->prev = &mmap_buf->next; + } + + *head = mmap_buf; + mmap_buf->prev = head; +} + + +nxt_inline void +nxt_unit_mmap_buf_insert_tail(nxt_unit_mmap_buf_t **prev, + nxt_unit_mmap_buf_t *mmap_buf) +{ + while (*prev != NULL) { + prev = &(*prev)->next; + } + + nxt_unit_mmap_buf_insert(prev, mmap_buf); +} + + +nxt_inline void +nxt_unit_mmap_buf_remove(nxt_unit_mmap_buf_t *mmap_buf) +{ + nxt_unit_mmap_buf_t **prev; + + prev = mmap_buf->prev; + + if (mmap_buf->next != NULL) { + mmap_buf->next->prev = prev; + } + + if (prev != NULL) { + *prev = mmap_buf->next; + } } @@ -509,26 +617,18 @@ int nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id, void *buf, size_t buf_size, void *oob, size_t oob_size) { - int fd, rc, nb; - pid_t pid; - nxt_queue_t incoming_buf; - struct cmsghdr *cm; - nxt_port_msg_t *port_msg; - nxt_unit_impl_t *lib; - nxt_unit_port_t new_port; - nxt_queue_link_t *lnk; - nxt_unit_request_t *r; - nxt_unit_mmap_buf_t *b; - nxt_unit_recv_msg_t recv_msg; - nxt_unit_callbacks_t *cb; - nxt_port_msg_new_port_t *new_port_msg; - nxt_unit_request_info_t *req; - nxt_unit_request_info_impl_t *req_impl; + int rc; + pid_t pid; + struct cmsghdr *cm; + nxt_port_msg_t *port_msg; + nxt_unit_impl_t *lib; + nxt_unit_recv_msg_t recv_msg; + nxt_unit_callbacks_t *cb; lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit); rc = NXT_UNIT_ERROR; - fd = -1; + recv_msg.fd = -1; recv_msg.process = NULL; port_msg = buf; cm = oob; @@ -538,17 +638,22 @@ nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id, && cm->cmsg_level == SOL_SOCKET && cm->cmsg_type == SCM_RIGHTS) { - memcpy(&fd, CMSG_DATA(cm), sizeof(int)); + memcpy(&recv_msg.fd, CMSG_DATA(cm), sizeof(int)); } - nxt_queue_init(&incoming_buf); + recv_msg.incoming_buf = NULL; if (nxt_slow_path(buf_size < sizeof(nxt_port_msg_t))) { nxt_unit_warn(ctx, "message too small (%d bytes)", (int) buf_size); goto fail; } - recv_msg.port_msg = *port_msg; + recv_msg.stream = port_msg->stream; + recv_msg.pid = port_msg->pid; + recv_msg.reply_port = port_msg->reply_port; + recv_msg.last = port_msg->last; + recv_msg.mmap = port_msg->mmap; + recv_msg.start = port_msg + 1; recv_msg.size = buf_size - sizeof(nxt_port_msg_t); @@ -572,7 +677,7 @@ nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id, } if (port_msg->mmap) { - if (nxt_unit_mmap_read(ctx, &recv_msg, &incoming_buf) != NXT_UNIT_OK) { + if (nxt_unit_mmap_read(ctx, &recv_msg) != NXT_UNIT_OK) { goto fail; } } @@ -589,187 +694,326 @@ nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id, break; case _NXT_PORT_MSG_NEW_PORT: - if (nxt_slow_path(recv_msg.size != sizeof(nxt_port_msg_new_port_t))) { - nxt_unit_warn(ctx, "#%"PRIu32": new_port: " - "invalid message size (%d)", - port_msg->stream, (int) recv_msg.size); + rc = nxt_unit_process_new_port(ctx, &recv_msg); + break; - goto fail; - } + case _NXT_PORT_MSG_CHANGE_FILE: + nxt_unit_debug(ctx, "#%"PRIu32": change_file: fd %d", + port_msg->stream, recv_msg.fd); + break; - if (nxt_slow_path(fd < 0)) { - nxt_unit_alert(ctx, "#%"PRIu32": invalid fd %d for new port", - port_msg->stream, fd); + case _NXT_PORT_MSG_MMAP: + if (nxt_slow_path(recv_msg.fd < 0)) { + nxt_unit_alert(ctx, "#%"PRIu32": invalid fd %d for mmap", + port_msg->stream, recv_msg.fd); goto fail; } - new_port_msg = recv_msg.start; + rc = nxt_unit_incoming_mmap(ctx, port_msg->pid, recv_msg.fd); + break; - nxt_unit_debug(ctx, "#%"PRIu32": new_port: %d,%d fd %d", - port_msg->stream, (int) new_port_msg->pid, - (int) new_port_msg->id, fd); + case _NXT_PORT_MSG_REQ_HEADERS: + rc = nxt_unit_process_req_headers(ctx, &recv_msg); + break; - nb = 0; + case _NXT_PORT_MSG_WEBSOCKET: + rc = nxt_unit_process_websocket(ctx, &recv_msg); + break; - if (nxt_slow_path(ioctl(fd, FIONBIO, &nb) == -1)) { - nxt_unit_alert(ctx, "#%"PRIu32": new_port: ioctl(%d, FIONBIO, 0) " - "failed: %s (%d)", fd, strerror(errno), errno); + case _NXT_PORT_MSG_REMOVE_PID: + if (nxt_slow_path(recv_msg.size != sizeof(pid))) { + nxt_unit_warn(ctx, "#%"PRIu32": remove_pid: invalid message size " + "(%d != %d)", port_msg->stream, (int) recv_msg.size, + (int) sizeof(pid)); goto fail; } - nxt_unit_port_id_init(&new_port.id, new_port_msg->pid, - new_port_msg->id); + memcpy(&pid, recv_msg.start, sizeof(pid)); - new_port.in_fd = -1; - new_port.out_fd = fd; - new_port.data = NULL; + nxt_unit_debug(ctx, "#%"PRIu32": remove_pid: %d", + port_msg->stream, (int) pid); - fd = -1; + cb->remove_pid(ctx, pid); - rc = cb->add_port(ctx, &new_port); + rc = NXT_UNIT_OK; break; - case _NXT_PORT_MSG_CHANGE_FILE: - nxt_unit_debug(ctx, "#%"PRIu32": change_file: fd %d", - port_msg->stream, fd); - break; + default: + nxt_unit_debug(ctx, "#%"PRIu32": ignore message type: %d", + port_msg->stream, (int) port_msg->type); - case _NXT_PORT_MSG_MMAP: - if (nxt_slow_path(fd < 0)) { - nxt_unit_alert(ctx, "#%"PRIu32": invalid fd %d for mmap", - port_msg->stream, fd); + goto fail; + } - goto fail; - } +fail: - rc = nxt_unit_incoming_mmap(ctx, port_msg->pid, fd); - break; + if (recv_msg.fd != -1) { + close(recv_msg.fd); + } - case _NXT_PORT_MSG_DATA: - if (nxt_slow_path(port_msg->mmap == 0)) { - nxt_unit_warn(ctx, "#%"PRIu32": data is not in shared memory", - port_msg->stream); + while (recv_msg.incoming_buf != NULL) { + nxt_unit_mmap_buf_free(recv_msg.incoming_buf); + } - goto fail; - } + if (recv_msg.process != NULL) { + nxt_unit_process_use(ctx, recv_msg.process, -1); + } - if (nxt_slow_path(recv_msg.size < sizeof(nxt_unit_request_t))) { - nxt_unit_warn(ctx, "#%"PRIu32": data too short: %d while at least " - "%d expected", port_msg->stream, (int) recv_msg.size, - (int) sizeof(nxt_unit_request_t)); + return rc; +} - goto fail; - } - req_impl = nxt_unit_request_info_get(ctx); - if (nxt_slow_path(req_impl == NULL)) { - nxt_unit_warn(ctx, "#%"PRIu32": request info allocation failed", - port_msg->stream); +static int +nxt_unit_process_new_port(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg) +{ + int nb; + nxt_unit_impl_t *lib; + nxt_unit_port_t new_port; + nxt_port_msg_new_port_t *new_port_msg; - goto fail; - } + if (nxt_slow_path(recv_msg->size != sizeof(nxt_port_msg_new_port_t))) { + nxt_unit_warn(ctx, "#%"PRIu32": new_port: " + "invalid message size (%d)", + recv_msg->stream, (int) recv_msg->size); - req = &req_impl->req; + return NXT_UNIT_ERROR; + } - req->request_port = *port_id; + if (nxt_slow_path(recv_msg->fd < 0)) { + nxt_unit_alert(ctx, "#%"PRIu32": invalid fd %d for new port", + recv_msg->stream, recv_msg->fd); - nxt_unit_port_id_init(&req->response_port, port_msg->pid, - port_msg->reply_port); + return NXT_UNIT_ERROR; + } - req->request = recv_msg.start; + new_port_msg = recv_msg->start; - lnk = nxt_queue_first(&incoming_buf); - b = nxt_container_of(lnk, nxt_unit_mmap_buf_t, link); + nxt_unit_debug(ctx, "#%"PRIu32": new_port: %d,%d fd %d", + recv_msg->stream, (int) new_port_msg->pid, + (int) new_port_msg->id, recv_msg->fd); - req->request_buf = &b->buf; - req->response = NULL; - req->response_buf = NULL; + nb = 0; - r = req->request; + if (nxt_slow_path(ioctl(recv_msg->fd, FIONBIO, &nb) == -1)) { + nxt_unit_alert(ctx, "#%"PRIu32": new_port: ioctl(%d, FIONBIO, 0) " + "failed: %s (%d)", recv_msg->fd, strerror(errno), errno); - req->content_length = r->content_length; + return NXT_UNIT_ERROR; + } - req->content_buf = req->request_buf; - req->content_buf->free = nxt_unit_sptr_get(&r->preread_content); + nxt_unit_port_id_init(&new_port.id, new_port_msg->pid, + new_port_msg->id); - /* Move process to req_impl. */ - req_impl->recv_msg = recv_msg; + new_port.in_fd = -1; + new_port.out_fd = recv_msg->fd; + new_port.data = NULL; - recv_msg.process = NULL; + recv_msg->fd = -1; - nxt_queue_init(&req_impl->outgoing_buf); - nxt_queue_init(&req_impl->incoming_buf); + lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit); - nxt_queue_each(b, &incoming_buf, nxt_unit_mmap_buf_t, link) - { - b->req = req; - } nxt_queue_loop; + return lib->callbacks.add_port(ctx, &new_port); +} - nxt_queue_add(&req_impl->incoming_buf, &incoming_buf); - nxt_queue_init(&incoming_buf); - req->response_max_fields = 0; - req_impl->state = NXT_UNIT_RS_START; +static int +nxt_unit_process_req_headers(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg) +{ + nxt_unit_impl_t *lib; + nxt_unit_request_t *r; + nxt_unit_mmap_buf_t *b; + nxt_unit_request_info_t *req; + nxt_unit_request_info_impl_t *req_impl; - nxt_unit_debug(ctx, "#%"PRIu32": %.*s %.*s (%d)", port_msg->stream, - (int) r->method_length, nxt_unit_sptr_get(&r->method), - (int) r->target_length, nxt_unit_sptr_get(&r->target), - (int) r->content_length); + if (nxt_slow_path(recv_msg->mmap == 0)) { + nxt_unit_warn(ctx, "#%"PRIu32": data is not in shared memory", + recv_msg->stream); - cb->request_handler(req); + return NXT_UNIT_ERROR; + } - rc = NXT_UNIT_OK; - break; + if (nxt_slow_path(recv_msg->size < sizeof(nxt_unit_request_t))) { + nxt_unit_warn(ctx, "#%"PRIu32": data too short: %d while at least " + "%d expected", recv_msg->stream, (int) recv_msg->size, + (int) sizeof(nxt_unit_request_t)); - case _NXT_PORT_MSG_REMOVE_PID: - if (nxt_slow_path(recv_msg.size != sizeof(pid))) { - nxt_unit_warn(ctx, "#%"PRIu32": remove_pid: invalid message size " - "(%d != %d)", port_msg->stream, (int) recv_msg.size, - (int) sizeof(pid)); + return NXT_UNIT_ERROR; + } - goto fail; - } + req_impl = nxt_unit_request_info_get(ctx); + if (nxt_slow_path(req_impl == NULL)) { + nxt_unit_warn(ctx, "#%"PRIu32": request info allocation failed", + recv_msg->stream); - memcpy(&pid, recv_msg.start, sizeof(pid)); + return NXT_UNIT_ERROR; + } - nxt_unit_debug(ctx, "#%"PRIu32": remove_pid: %d", - port_msg->stream, (int) pid); + req = &req_impl->req; - cb->remove_pid(ctx, pid); + nxt_unit_port_id_init(&req->response_port, recv_msg->pid, + recv_msg->reply_port); - rc = NXT_UNIT_OK; - break; + req->request = recv_msg->start; - default: - nxt_unit_debug(ctx, "#%"PRIu32": ignore message type: %d", - port_msg->stream, (int) port_msg->type); + b = recv_msg->incoming_buf; - goto fail; + req->request_buf = &b->buf; + req->response = NULL; + req->response_buf = NULL; + + r = req->request; + + req->content_length = r->content_length; + + req->content_buf = req->request_buf; + req->content_buf->free = nxt_unit_sptr_get(&r->preread_content); + + /* "Move" process reference to req_impl. */ + req_impl->process = nxt_unit_msg_get_process(ctx, recv_msg); + if (nxt_slow_path(req_impl->process == NULL)) { + return NXT_UNIT_ERROR; } -fail: + recv_msg->process = NULL; - if (fd != -1) { - close(fd); + req_impl->stream = recv_msg->stream; + + req_impl->outgoing_buf = NULL; + + for (b = recv_msg->incoming_buf; b != NULL; b = b->next) { + b->req = req; } - if (port_msg->mmap) { - nxt_queue_each(b, &incoming_buf, nxt_unit_mmap_buf_t, link) - { - nxt_unit_mmap_release(b->hdr, b->buf.start, - b->buf.end - b->buf.start); + /* "Move" incoming buffer list to req_impl. */ + req_impl->incoming_buf = recv_msg->incoming_buf; + req_impl->incoming_buf->prev = &req_impl->incoming_buf; + recv_msg->incoming_buf = NULL; + + req->response_max_fields = 0; + req_impl->state = NXT_UNIT_RS_START; + req_impl->websocket = 0; + + nxt_unit_debug(ctx, "#%"PRIu32": %.*s %.*s (%d)", recv_msg->stream, + (int) r->method_length, nxt_unit_sptr_get(&r->method), + (int) r->target_length, nxt_unit_sptr_get(&r->target), + (int) r->content_length); + + lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit); + + lib->callbacks.request_handler(req); - nxt_unit_mmap_buf_release(b); - } nxt_queue_loop; + return NXT_UNIT_OK; +} + + +static int +nxt_unit_process_websocket(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg) +{ + size_t hsize; + nxt_unit_impl_t *lib; + nxt_unit_mmap_buf_t *b; + nxt_unit_ctx_impl_t *ctx_impl; + nxt_unit_callbacks_t *cb; + nxt_unit_request_info_t *req; + nxt_unit_request_info_impl_t *req_impl; + nxt_unit_websocket_frame_impl_t *ws_impl; + + ctx_impl = nxt_container_of(ctx, nxt_unit_ctx_impl_t, ctx); + + req_impl = nxt_unit_request_hash_find(&ctx_impl->requests, recv_msg->stream, + recv_msg->last); + if (req_impl == NULL) { + return NXT_UNIT_OK; } - if (recv_msg.process != NULL) { - nxt_unit_process_use(ctx, recv_msg.process, -1); + req = &req_impl->req; + + lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit); + cb = &lib->callbacks; + + if (cb->websocket_handler && recv_msg->size >= 2) { + ws_impl = nxt_unit_websocket_frame_get(ctx); + if (nxt_slow_path(ws_impl == NULL)) { + nxt_unit_warn(ctx, "#%"PRIu32": websocket frame allocation failed", + req_impl->stream); + + return NXT_UNIT_ERROR; + } + + ws_impl->ws.req = req; + + ws_impl->buf = NULL; + ws_impl->retain_buf = NULL; + + if (recv_msg->mmap) { + for (b = recv_msg->incoming_buf; b != NULL; b = b->next) { + b->req = req; + } + + /* "Move" incoming buffer list to ws_impl. */ + ws_impl->buf = recv_msg->incoming_buf; + ws_impl->buf->prev = &ws_impl->buf; + recv_msg->incoming_buf = NULL; + + b = ws_impl->buf; + + } else { + b = nxt_unit_mmap_buf_get(ctx); + if (nxt_slow_path(b == NULL)) { + return NXT_UNIT_ERROR; + } + + b->hdr = NULL; + b->req = req; + b->buf.start = recv_msg->start; + b->buf.free = b->buf.start; + b->buf.end = b->buf.start + recv_msg->size; + + nxt_unit_mmap_buf_insert(&ws_impl->buf, b); + } + + ws_impl->ws.header = (void *) b->buf.start; + ws_impl->ws.payload_len = nxt_websocket_frame_payload_len( + ws_impl->ws.header); + + hsize = nxt_websocket_frame_header_size(ws_impl->ws.header); + + if (ws_impl->ws.header->mask) { + ws_impl->ws.mask = (uint8_t *) b->buf.start + hsize - 4; + + } else { + ws_impl->ws.mask = NULL; + } + + b->buf.free += hsize; + + ws_impl->ws.content_buf = &b->buf; + ws_impl->ws.content_length = ws_impl->ws.payload_len; + + nxt_unit_req_debug(req, "websocket_handler: opcode=%d, " + "payload_len=%"PRIu64, + ws_impl->ws.header->opcode, + ws_impl->ws.payload_len); + + cb->websocket_handler(&ws_impl->ws); } - return rc; + if (recv_msg->last) { + req_impl->websocket = 0; + + if (cb->close_handler) { + nxt_unit_req_debug(req, "close_handler"); + + cb->close_handler(req); + + } else { + nxt_unit_request_done(req, NXT_UNIT_ERROR); + } + } + + return NXT_UNIT_OK; } @@ -815,9 +1059,7 @@ nxt_unit_request_info_get(nxt_unit_ctx_t *ctx) static void nxt_unit_request_info_release(nxt_unit_request_info_t *req) { - nxt_unit_mmap_buf_t *b; nxt_unit_ctx_impl_t *ctx_impl; - nxt_unit_recv_msg_t *recv_msg; nxt_unit_request_info_impl_t *req_impl; ctx_impl = nxt_container_of(req->ctx, nxt_unit_ctx_impl_t, ctx); @@ -826,30 +1068,31 @@ nxt_unit_request_info_release(nxt_unit_request_info_t *req) req->response = NULL; req->response_buf = NULL; - recv_msg = &req_impl->recv_msg; + if (req_impl->process != NULL) { + nxt_unit_process_use(req->ctx, req_impl->process, -1); - if (recv_msg->process != NULL) { - nxt_unit_process_use(req->ctx, recv_msg->process, -1); - - recv_msg->process = NULL; + req_impl->process = NULL; } - nxt_queue_each(b, &req_impl->outgoing_buf, nxt_unit_mmap_buf_t, link) { - - nxt_unit_buf_free(&b->buf); + if (req_impl->websocket) { + nxt_unit_request_hash_find(&ctx_impl->requests, req_impl->stream, 1); - } nxt_queue_loop; - - nxt_queue_each(b, &req_impl->incoming_buf, nxt_unit_mmap_buf_t, link) { + req_impl->websocket = 0; + } - nxt_unit_mmap_release(b->hdr, b->buf.start, b->buf.end - b->buf.start); - nxt_unit_mmap_buf_release(b); + while (req_impl->outgoing_buf != NULL) { + nxt_unit_mmap_buf_free(req_impl->outgoing_buf); + } - } nxt_queue_loop; + while (req_impl->incoming_buf != NULL) { + nxt_unit_mmap_buf_free(req_impl->incoming_buf); + } nxt_queue_remove(&req_impl->link); nxt_queue_insert_tail(&ctx_impl->free_req, &req_impl->link); + + req_impl->state = NXT_UNIT_RS_RELEASED; } @@ -868,6 +1111,68 @@ nxt_unit_request_info_free(nxt_unit_request_info_impl_t *req_impl) } +static nxt_unit_websocket_frame_impl_t * +nxt_unit_websocket_frame_get(nxt_unit_ctx_t *ctx) +{ + nxt_queue_link_t *lnk; + nxt_unit_ctx_impl_t *ctx_impl; + nxt_unit_websocket_frame_impl_t *ws_impl; + + ctx_impl = nxt_container_of(ctx, nxt_unit_ctx_impl_t, ctx); + + if (nxt_queue_is_empty(&ctx_impl->free_ws)) { + ws_impl = malloc(sizeof(nxt_unit_websocket_frame_impl_t)); + if (nxt_slow_path(ws_impl == NULL)) { + nxt_unit_warn(ctx, "websocket frame allocation failed"); + + return NULL; + } + + } else { + lnk = nxt_queue_first(&ctx_impl->free_ws); + nxt_queue_remove(lnk); + + ws_impl = nxt_container_of(lnk, nxt_unit_websocket_frame_impl_t, link); + } + + ws_impl->ctx_impl = ctx_impl; + + return ws_impl; +} + + +static void +nxt_unit_websocket_frame_release(nxt_unit_websocket_frame_t *ws) +{ + nxt_unit_websocket_frame_impl_t *ws_impl; + + ws_impl = nxt_container_of(ws, nxt_unit_websocket_frame_impl_t, ws); + + while (ws_impl->buf != NULL) { + nxt_unit_mmap_buf_free(ws_impl->buf); + } + + ws->req = NULL; + + if (ws_impl->retain_buf != NULL) { + free(ws_impl->retain_buf); + + ws_impl->retain_buf = NULL; + } + + nxt_queue_insert_tail(&ws_impl->ctx_impl->free_ws, &ws_impl->link); +} + + +static void +nxt_unit_websocket_frame_free(nxt_unit_websocket_frame_impl_t *ws_impl) +{ + nxt_queue_remove(&ws_impl->link); + + free(ws_impl); +} + + uint16_t nxt_unit_field_hash(const char *name, size_t name_length) { @@ -1275,6 +1580,10 @@ nxt_unit_response_send(nxt_unit_request_info_t *req) return NXT_UNIT_ERROR; } + if (req->request->websocket_handshake && req->response->status == 101) { + nxt_unit_response_upgrade(req); + } + nxt_unit_req_debug(req, "send: %"PRIu32" fields, %d bytes", req->response->fields_count, (int) (req->response_buf->free @@ -1282,9 +1591,7 @@ nxt_unit_response_send(nxt_unit_request_info_t *req) mmap_buf = nxt_container_of(req->response_buf, nxt_unit_mmap_buf_t, buf); - rc = nxt_unit_mmap_buf_send(req->ctx, - req_impl->recv_msg.port_msg.stream, - mmap_buf, 0); + rc = nxt_unit_mmap_buf_send(req->ctx, req_impl->stream, mmap_buf, 0); if (nxt_fast_path(rc == NXT_UNIT_OK)) { req->response = NULL; req->response_buf = NULL; @@ -1312,7 +1619,6 @@ nxt_unit_buf_t * nxt_unit_response_buf_alloc(nxt_unit_request_info_t *req, uint32_t size) { int rc; - nxt_unit_process_t *process; nxt_unit_mmap_buf_t *mmap_buf; nxt_unit_request_info_impl_t *req_impl; @@ -1327,11 +1633,6 @@ nxt_unit_response_buf_alloc(nxt_unit_request_info_t *req, uint32_t size) req_impl = nxt_container_of(req, nxt_unit_request_info_impl_t, req); - process = nxt_unit_msg_get_process(req->ctx, &req_impl->recv_msg); - if (nxt_slow_path(process == NULL)) { - return NULL; - } - mmap_buf = nxt_unit_mmap_buf_get(req->ctx); if (nxt_slow_path(mmap_buf == NULL)) { return NULL; @@ -1339,10 +1640,10 @@ nxt_unit_response_buf_alloc(nxt_unit_request_info_t *req, uint32_t size) mmap_buf->req = req; - nxt_queue_insert_tail(&req_impl->outgoing_buf, &mmap_buf->link); + nxt_unit_mmap_buf_insert_tail(&req_impl->outgoing_buf, mmap_buf); - rc = nxt_unit_get_outgoing_buf(req->ctx, process, &req->response_port, - size, mmap_buf); + rc = nxt_unit_get_outgoing_buf(req->ctx, req_impl->process, + &req->response_port, size, mmap_buf); if (nxt_slow_path(rc != NXT_UNIT_OK)) { nxt_unit_mmap_buf_release(mmap_buf); @@ -1366,13 +1667,13 @@ nxt_unit_msg_get_process(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg) pthread_mutex_lock(&lib->mutex); - recv_msg->process = nxt_unit_process_find(ctx, recv_msg->port_msg.pid, 0); + recv_msg->process = nxt_unit_process_find(ctx, recv_msg->pid, 0); pthread_mutex_unlock(&lib->mutex); if (recv_msg->process == NULL) { nxt_unit_warn(ctx, "#%"PRIu32": process %d not found", - recv_msg->port_msg.stream, (int) recv_msg->port_msg.pid); + recv_msg->stream, (int) recv_msg->pid); } return recv_msg->process; @@ -1382,23 +1683,21 @@ nxt_unit_msg_get_process(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg) static nxt_unit_mmap_buf_t * nxt_unit_mmap_buf_get(nxt_unit_ctx_t *ctx) { - nxt_queue_link_t *lnk; nxt_unit_mmap_buf_t *mmap_buf; nxt_unit_ctx_impl_t *ctx_impl; ctx_impl = nxt_container_of(ctx, nxt_unit_ctx_impl_t, ctx); - if (nxt_queue_is_empty(&ctx_impl->free_buf)) { + if (ctx_impl->free_buf == NULL) { mmap_buf = malloc(sizeof(nxt_unit_mmap_buf_t)); if (nxt_slow_path(mmap_buf == NULL)) { nxt_unit_warn(ctx, "failed to allocate buf"); } } else { - lnk = nxt_queue_first(&ctx_impl->free_buf); - nxt_queue_remove(lnk); + mmap_buf = ctx_impl->free_buf; - mmap_buf = nxt_container_of(lnk, nxt_unit_mmap_buf_t, link); + nxt_unit_mmap_buf_remove(mmap_buf); } mmap_buf->ctx_impl = ctx_impl; @@ -1410,9 +1709,91 @@ nxt_unit_mmap_buf_get(nxt_unit_ctx_t *ctx) static void nxt_unit_mmap_buf_release(nxt_unit_mmap_buf_t *mmap_buf) { - nxt_queue_remove(&mmap_buf->link); + nxt_unit_mmap_buf_remove(mmap_buf); + + nxt_unit_mmap_buf_insert(&mmap_buf->ctx_impl->free_buf, mmap_buf); +} + + +typedef struct { + size_t len; + const char *str; +} nxt_unit_str_t; + + +#define nxt_unit_str(str) { nxt_length(str), str } + + +int +nxt_unit_request_is_websocket_handshake(nxt_unit_request_info_t *req) +{ + return req->request->websocket_handshake; +} + + +int +nxt_unit_response_upgrade(nxt_unit_request_info_t *req) +{ + int rc; + nxt_unit_ctx_impl_t *ctx_impl; + nxt_unit_request_info_impl_t *req_impl; + + req_impl = nxt_container_of(req, nxt_unit_request_info_impl_t, req); + + if (nxt_slow_path(req_impl->websocket != 0)) { + nxt_unit_req_debug(req, "upgrade: already upgraded"); + + return NXT_UNIT_OK; + } + + if (nxt_slow_path(req_impl->state < NXT_UNIT_RS_RESPONSE_INIT)) { + nxt_unit_req_warn(req, "upgrade: response is not initialized yet"); + + return NXT_UNIT_ERROR; + } + + if (nxt_slow_path(req_impl->state >= NXT_UNIT_RS_RESPONSE_SENT)) { + nxt_unit_req_warn(req, "upgrade: response already sent"); + + return NXT_UNIT_ERROR; + } + + ctx_impl = nxt_container_of(req->ctx, nxt_unit_ctx_impl_t, ctx); + + rc = nxt_unit_request_hash_add(&ctx_impl->requests, req_impl); + if (nxt_slow_path(rc != NXT_UNIT_OK)) { + nxt_unit_req_warn(req, "upgrade: failed to add request to hash"); + + return NXT_UNIT_ERROR; + } + + req_impl->websocket = 1; + + req->response->status = 101; + + return NXT_UNIT_OK; +} + + +int +nxt_unit_response_is_websocket(nxt_unit_request_info_t *req) +{ + nxt_unit_request_info_impl_t *req_impl; - nxt_queue_insert_tail(&mmap_buf->ctx_impl->free_buf, &mmap_buf->link); + req_impl = nxt_container_of(req, nxt_unit_request_info_impl_t, req); + + return req_impl->websocket; +} + + +nxt_unit_request_info_t * +nxt_unit_get_request_info_from_data(void *data) +{ + nxt_unit_request_info_impl_t *req_impl; + + req_impl = nxt_container_of(data, nxt_unit_request_info_impl_t, extra_data); + + return &req_impl->req; } @@ -1445,9 +1826,7 @@ nxt_unit_buf_send(nxt_unit_buf_t *buf) } if (nxt_fast_path(buf->free > buf->start)) { - rc = nxt_unit_mmap_buf_send(req->ctx, - req_impl->recv_msg.port_msg.stream, - mmap_buf, 0); + rc = nxt_unit_mmap_buf_send(req->ctx, req_impl->stream, mmap_buf, 0); if (nxt_slow_path(rc != NXT_UNIT_OK)) { return rc; } @@ -1472,10 +1851,7 @@ nxt_unit_buf_send_done(nxt_unit_buf_t *buf) req = mmap_buf->req; req_impl = nxt_container_of(req, nxt_unit_request_info_impl_t, req); - rc = nxt_unit_mmap_buf_send(req->ctx, - req_impl->recv_msg.port_msg.stream, - mmap_buf, 1); - + rc = nxt_unit_mmap_buf_send(req->ctx, req_impl->stream, mmap_buf, 1); if (nxt_slow_path(rc == NXT_UNIT_OK)) { nxt_unit_mmap_buf_release(mmap_buf); @@ -1506,6 +1882,7 @@ nxt_unit_mmap_buf_send(nxt_unit_ctx_t *ctx, uint32_t stream, lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit); buf = &mmap_buf->buf; + hdr = mmap_buf->hdr; m.mmap_msg.size = buf->free - buf->start; @@ -1514,15 +1891,15 @@ nxt_unit_mmap_buf_send(nxt_unit_ctx_t *ctx, uint32_t stream, m.msg.reply_port = 0; m.msg.type = _NXT_PORT_MSG_DATA; m.msg.last = last != 0; - m.msg.mmap = m.mmap_msg.size > 0; + m.msg.mmap = hdr != NULL && m.mmap_msg.size > 0; m.msg.nf = 0; m.msg.mf = 0; m.msg.tracking = 0; - hdr = mmap_buf->hdr; - - m.mmap_msg.mmap_id = hdr->id; - m.mmap_msg.chunk_id = nxt_port_mmap_chunk_id(hdr, (u_char *) buf->start); + if (hdr != NULL) { + m.mmap_msg.mmap_id = hdr->id; + m.mmap_msg.chunk_id = nxt_port_mmap_chunk_id(hdr, (u_char *) buf->start); + } nxt_unit_debug(ctx, "#%"PRIu32": send mmap: (%d,%d,%d)", stream, @@ -1531,14 +1908,13 @@ nxt_unit_mmap_buf_send(nxt_unit_ctx_t *ctx, uint32_t stream, (int) m.mmap_msg.size); res = lib->callbacks.port_send(ctx, &mmap_buf->port_id, &m, - m.mmap_msg.size > 0 ? sizeof(m) - : sizeof(m.msg), + m.msg.mmap ? sizeof(m) : sizeof(m.msg), NULL, 0); if (nxt_slow_path(res != sizeof(m))) { return NXT_UNIT_ERROR; } - if (buf->end - buf->free >= PORT_MMAP_CHUNK_SIZE) { + if (buf->end - buf->free >= PORT_MMAP_CHUNK_SIZE && hdr != NULL) { last_used = (u_char *) buf->free - 1; first_free_chunk = nxt_port_mmap_chunk_id(hdr, last_used) + 1; @@ -1557,11 +1933,17 @@ nxt_unit_mmap_buf_send(nxt_unit_ctx_t *ctx, uint32_t stream, void nxt_unit_buf_free(nxt_unit_buf_t *buf) { - nxt_unit_mmap_buf_t *mmap_buf; + nxt_unit_mmap_buf_free(nxt_container_of(buf, nxt_unit_mmap_buf_t, buf)); +} - mmap_buf = nxt_container_of(buf, nxt_unit_mmap_buf_t, buf); - nxt_unit_mmap_release(mmap_buf->hdr, buf->start, buf->end - buf->start); +static void +nxt_unit_mmap_buf_free(nxt_unit_mmap_buf_t *mmap_buf) +{ + if (nxt_fast_path(mmap_buf->hdr != NULL)) { + nxt_unit_mmap_release(mmap_buf->hdr, mmap_buf->buf.start, + mmap_buf->buf.end - mmap_buf->buf.start); + } nxt_unit_mmap_buf_release(mmap_buf); } @@ -1570,26 +1952,15 @@ nxt_unit_buf_free(nxt_unit_buf_t *buf) nxt_unit_buf_t * nxt_unit_buf_next(nxt_unit_buf_t *buf) { - nxt_queue_link_t *lnk; - nxt_unit_mmap_buf_t *mmap_buf; - nxt_unit_request_info_impl_t *req_impl; + nxt_unit_mmap_buf_t *mmap_buf; mmap_buf = nxt_container_of(buf, nxt_unit_mmap_buf_t, buf); - req_impl = nxt_container_of(mmap_buf->req, nxt_unit_request_info_impl_t, - req); - lnk = &mmap_buf->link; - - if (lnk == nxt_queue_last(&req_impl->incoming_buf) - || lnk == nxt_queue_last(&req_impl->outgoing_buf)) - { + if (mmap_buf->next == NULL) { return NULL; } - lnk = nxt_queue_next(lnk); - mmap_buf = nxt_container_of(lnk, nxt_unit_mmap_buf_t, link); - - return &mmap_buf->buf; + return &mmap_buf->next->buf; } @@ -1614,7 +1985,6 @@ nxt_unit_response_write(nxt_unit_request_info_t *req, const void *start, int rc; uint32_t part_size; const char *part_start; - nxt_unit_process_t *process; nxt_unit_mmap_buf_t mmap_buf; nxt_unit_request_info_impl_t *req_impl; @@ -1641,16 +2011,12 @@ nxt_unit_response_write(nxt_unit_request_info_t *req, const void *start, part_start += part_size; } - process = nxt_unit_msg_get_process(req->ctx, &req_impl->recv_msg); - if (nxt_slow_path(process == NULL)) { - return NXT_UNIT_ERROR; - } - while (size > 0) { part_size = nxt_min(size, PORT_MMAP_DATA_SIZE); - rc = nxt_unit_get_outgoing_buf(req->ctx, process, &req->response_port, - part_size, &mmap_buf); + rc = nxt_unit_get_outgoing_buf(req->ctx, req_impl->process, + &req->response_port, part_size, + &mmap_buf); if (nxt_slow_path(rc != NXT_UNIT_OK)) { return rc; } @@ -1658,9 +2024,7 @@ nxt_unit_response_write(nxt_unit_request_info_t *req, const void *start, mmap_buf.buf.free = nxt_cpymem(mmap_buf.buf.free, part_start, part_size); - rc = nxt_unit_mmap_buf_send(req->ctx, - req_impl->recv_msg.port_msg.stream, - &mmap_buf, 0); + rc = nxt_unit_mmap_buf_send(req->ctx, req_impl->stream, &mmap_buf, 0); if (nxt_slow_path(rc != NXT_UNIT_OK)) { nxt_unit_mmap_release(mmap_buf.hdr, mmap_buf.buf.start, mmap_buf.buf.end - mmap_buf.buf.start); @@ -1766,6 +2130,14 @@ nxt_unit_response_write_cb(nxt_unit_request_info_t *req, ssize_t nxt_unit_request_read(nxt_unit_request_info_t *req, void *dst, size_t size) { + return nxt_unit_buf_read(&req->content_buf, &req->content_length, + dst, size); +} + + +static ssize_t +nxt_unit_buf_read(nxt_unit_buf_t **b, uint64_t *len, void *dst, size_t size) +{ u_char *p; size_t rest, copy, read; nxt_unit_buf_t *buf; @@ -1773,7 +2145,7 @@ nxt_unit_request_read(nxt_unit_request_info_t *req, void *dst, size_t size) p = dst; rest = size; - buf = req->content_buf; + buf = *b; while (buf != NULL) { copy = buf->end - buf->free; @@ -1795,11 +2167,11 @@ nxt_unit_request_read(nxt_unit_request_info_t *req, void *dst, size_t size) buf = nxt_unit_buf_next(buf); } - req->content_buf = buf; + *b = buf; read = size - rest; - req->content_length -= read; + *len -= read; return read; } @@ -1852,7 +2224,7 @@ skip_response_send: lib = nxt_container_of(req->unit, nxt_unit_impl_t, unit); - msg.stream = req_impl->recv_msg.port_msg.stream; + msg.stream = req_impl->stream; msg.pid = lib->pid; msg.reply_port = 0; msg.type = (rc == NXT_UNIT_OK) ? _NXT_PORT_MSG_DATA @@ -1874,6 +2246,162 @@ skip_response_send: } +int +nxt_unit_websocket_send(nxt_unit_request_info_t *req, uint8_t opcode, + uint8_t last, const void *start, size_t size) +{ + const struct iovec iov = { (void *) start, size }; + + return nxt_unit_websocket_sendv(req, opcode, last, &iov, 1); +} + + +int +nxt_unit_websocket_sendv(nxt_unit_request_info_t *req, uint8_t opcode, + uint8_t last, const struct iovec *iov, int iovcnt) +{ + int i, rc; + size_t l, copy; + uint32_t payload_len, buf_size; + const uint8_t *b; + nxt_unit_buf_t *buf; + nxt_websocket_header_t *wh; + + payload_len = 0; + + for (i = 0; i < iovcnt; i++) { + payload_len += iov[i].iov_len; + } + + buf_size = 10 + payload_len; + + buf = nxt_unit_response_buf_alloc(req, nxt_min(buf_size, + PORT_MMAP_DATA_SIZE)); + if (nxt_slow_path(buf == NULL)) { + nxt_unit_req_error(req, "Failed to allocate buf for content"); + + return NXT_UNIT_ERROR; + } + + buf->start[0] = 0; + buf->start[1] = 0; + + wh = (void *) buf->free; + + buf->free = nxt_websocket_frame_init(wh, payload_len); + wh->fin = last; + wh->opcode = opcode; + + for (i = 0; i < iovcnt; i++) { + b = iov[i].iov_base; + l = iov[i].iov_len; + + while (l > 0) { + copy = buf->end - buf->free; + copy = nxt_min(l, copy); + + buf->free = nxt_cpymem(buf->free, b, copy); + b += copy; + l -= copy; + + if (l > 0) { + buf_size -= buf->end - buf->start; + + rc = nxt_unit_buf_send(buf); + if (nxt_slow_path(rc != NXT_UNIT_OK)) { + nxt_unit_req_error(req, "Failed to send content"); + + return NXT_UNIT_ERROR; + } + + buf = nxt_unit_response_buf_alloc(req, nxt_min(buf_size, + PORT_MMAP_DATA_SIZE)); + if (nxt_slow_path(buf == NULL)) { + nxt_unit_req_error(req, + "Failed to allocate buf for content"); + + return NXT_UNIT_ERROR; + } + } + } + } + + if (buf->free > buf->start) { + rc = nxt_unit_buf_send(buf); + if (nxt_slow_path(rc != NXT_UNIT_OK)) { + nxt_unit_req_error(req, "Failed to send content"); + } + } + + return rc; +} + + +ssize_t +nxt_unit_websocket_read(nxt_unit_websocket_frame_t *ws, void *dst, + size_t size) +{ + ssize_t res; + uint8_t *b; + uint64_t i, d; + + res = nxt_unit_buf_read(&ws->content_buf, &ws->content_length, + dst, size); + + if (ws->mask == NULL) { + return res; + } + + b = dst; + d = (ws->payload_len - ws->content_length - res) % 4; + + for (i = 0; i < (uint64_t) res; i++) { + b[i] ^= ws->mask[ (i + d) % 4 ]; + } + + return res; +} + + +int +nxt_unit_websocket_retain(nxt_unit_websocket_frame_t *ws) +{ + char *b; + size_t size; + nxt_unit_websocket_frame_impl_t *ws_impl; + + ws_impl = nxt_container_of(ws, nxt_unit_websocket_frame_impl_t, ws); + + if (ws_impl->retain_buf != NULL || ws_impl->buf->hdr != NULL) { + return NXT_UNIT_OK; + } + + size = ws_impl->buf->buf.end - ws_impl->buf->buf.start; + + b = malloc(size); + if (nxt_slow_path(b == NULL)) { + return NXT_UNIT_ERROR; + } + + memcpy(b, ws_impl->buf->buf.start, size); + + ws_impl->buf->buf.start = b; + ws_impl->buf->buf.free = b; + ws_impl->buf->buf.end = b + size; + + ws_impl->retain_buf = b; + + return NXT_UNIT_OK; +} + + +void +nxt_unit_websocket_done(nxt_unit_websocket_frame_t *ws) +{ + nxt_unit_websocket_frame_release(ws); +} + + static nxt_port_mmap_header_t * nxt_unit_mmap_get(nxt_unit_ctx_t *ctx, nxt_unit_process_t *process, nxt_unit_port_id_t *port_id, nxt_chunk_id_t *c, int n) @@ -2355,7 +2883,7 @@ nxt_unit_tracking_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg) if (recv_msg->size < (int) sizeof(nxt_port_mmap_tracking_msg_t)) { nxt_unit_warn(ctx, "#%"PRIu32": tracking_read: too small message (%d)", - recv_msg->port_msg.stream, (int) recv_msg->size); + recv_msg->stream, (int) recv_msg->size); return 0; } @@ -2378,18 +2906,18 @@ nxt_unit_tracking_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg) nxt_unit_warn(ctx, "#%"PRIu32": tracking_read: " "invalid mmap id %d,%"PRIu32, - recv_msg->port_msg.stream, - (int) process->pid, tracking_msg->mmap_id); + recv_msg->stream, (int) process->pid, + tracking_msg->mmap_id); return 0; } c = tracking_msg->tracking_id; - rc = nxt_atomic_cmp_set(hdr->tracking + c, recv_msg->port_msg.stream, 0); + rc = nxt_atomic_cmp_set(hdr->tracking + c, recv_msg->stream, 0); if (rc == 0) { nxt_unit_debug(ctx, "#%"PRIu32": tracking cancelled", - recv_msg->port_msg.stream); + recv_msg->stream); nxt_port_mmap_set_chunk_free(hdr->free_tracking_map, c); } @@ -2401,19 +2929,18 @@ nxt_unit_tracking_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg) static int -nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg, - nxt_queue_t *incoming_buf) +nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg) { void *start; uint32_t size; nxt_unit_process_t *process; - nxt_unit_mmap_buf_t *b; + nxt_unit_mmap_buf_t *b, **incoming_tail; nxt_port_mmap_msg_t *mmap_msg, *end; nxt_port_mmap_header_t *hdr; if (nxt_slow_path(recv_msg->size < sizeof(nxt_port_mmap_msg_t))) { nxt_unit_warn(ctx, "#%"PRIu32": mmap_read: too small message (%d)", - recv_msg->port_msg.stream, (int) recv_msg->size); + recv_msg->stream, (int) recv_msg->size); return NXT_UNIT_ERROR; } @@ -2426,6 +2953,8 @@ nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg, mmap_msg = recv_msg->start; end = nxt_pointer_to(recv_msg->start, recv_msg->size); + incoming_tail = &recv_msg->incoming_buf; + pthread_mutex_lock(&process->incoming.mutex); for (; mmap_msg < end; mmap_msg++) { @@ -2435,8 +2964,8 @@ nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg, nxt_unit_warn(ctx, "#%"PRIu32": mmap_read: " "invalid mmap id %d,%"PRIu32, - recv_msg->port_msg.stream, - (int) process->pid, mmap_msg->mmap_id); + recv_msg->stream, (int) process->pid, + mmap_msg->mmap_id); return NXT_UNIT_ERROR; } @@ -2453,16 +2982,16 @@ nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg, if (nxt_slow_path(b == NULL)) { pthread_mutex_unlock(&process->incoming.mutex); - nxt_unit_warn(ctx, "#%"PRIu32": mmap_read: " - "failed to allocate buf", - recv_msg->port_msg.stream); + nxt_unit_warn(ctx, "#%"PRIu32": mmap_read: failed to allocate buf", + recv_msg->stream); nxt_unit_mmap_release(hdr, start, size); return NXT_UNIT_ERROR; } - nxt_queue_insert_tail(incoming_buf, &b->link); + nxt_unit_mmap_buf_insert(incoming_tail, b); + incoming_tail = &b->next; b->buf.start = start; b->buf.free = start; @@ -2470,7 +2999,7 @@ nxt_unit_mmap_read(nxt_unit_ctx_t *ctx, nxt_unit_recv_msg_t *recv_msg, b->hdr = hdr; nxt_unit_debug(ctx, "#%"PRIu32": mmap_read: [%p,%d] %d->%d,(%d,%d,%d)", - recv_msg->port_msg.stream, + recv_msg->stream, start, (int) size, (int) hdr->src_pid, (int) hdr->dst_pid, (int) hdr->id, (int) mmap_msg->chunk_id, @@ -2685,6 +3214,11 @@ nxt_unit_run_once(nxt_unit_ctx_t *ctx) if (nxt_fast_path(rsize > 0)) { rc = nxt_unit_process_msg(ctx, &ctx_impl->read_port_id, buf, rsize, oob, sizeof(oob)); + +#if (NXT_DEBUG) + memset(buf, 0xAC, rsize); +#endif + } else { rc = NXT_UNIT_ERROR; } @@ -2775,10 +3309,11 @@ nxt_unit_ctx_alloc(nxt_unit_ctx_t *ctx, void *data) void nxt_unit_ctx_free(nxt_unit_ctx_t *ctx) { - nxt_unit_impl_t *lib; - nxt_unit_ctx_impl_t *ctx_impl; - nxt_unit_mmap_buf_t *mmap_buf; - nxt_unit_request_info_impl_t *req_impl; + nxt_unit_impl_t *lib; + nxt_unit_ctx_impl_t *ctx_impl; + nxt_unit_mmap_buf_t *mmap_buf; + nxt_unit_request_info_impl_t *req_impl; + nxt_unit_websocket_frame_impl_t *ws_impl; ctx_impl = nxt_container_of(ctx, nxt_unit_ctx_impl_t, ctx); lib = nxt_container_of(ctx->unit, nxt_unit_impl_t, unit); @@ -2792,15 +3327,14 @@ nxt_unit_ctx_free(nxt_unit_ctx_t *ctx) } nxt_queue_loop; - nxt_queue_remove(&ctx_impl->ctx_buf[0].link); - nxt_queue_remove(&ctx_impl->ctx_buf[1].link); - - nxt_queue_each(mmap_buf, &ctx_impl->free_buf, nxt_unit_mmap_buf_t, link) { + nxt_unit_mmap_buf_remove(&ctx_impl->ctx_buf[0]); + nxt_unit_mmap_buf_remove(&ctx_impl->ctx_buf[1]); - nxt_queue_remove(&mmap_buf->link); + while (ctx_impl->free_buf != NULL) { + mmap_buf = ctx_impl->free_buf; + nxt_unit_mmap_buf_remove(mmap_buf); free(mmap_buf); - - } nxt_queue_loop; + } nxt_queue_each(req_impl, &ctx_impl->free_req, nxt_unit_request_info_impl_t, link) @@ -2809,6 +3343,13 @@ nxt_unit_ctx_free(nxt_unit_ctx_t *ctx) } nxt_queue_loop; + nxt_queue_each(ws_impl, &ctx_impl->free_ws, + nxt_unit_websocket_frame_impl_t, link) + { + nxt_unit_websocket_frame_free(ws_impl); + + } nxt_queue_loop; + nxt_queue_remove(&ctx_impl->link); if (ctx_impl != &lib->main_ctx) { @@ -3454,6 +3995,83 @@ nxt_unit_port_hash_find(nxt_lvlhsh_t *port_hash, nxt_unit_port_id_t *port_id, } +static nxt_int_t +nxt_unit_request_hash_test(nxt_lvlhsh_query_t *lhq, void *data) +{ + return NXT_OK; +} + + +static const nxt_lvlhsh_proto_t lvlhsh_requests_proto nxt_aligned(64) = { + NXT_LVLHSH_DEFAULT, + nxt_unit_request_hash_test, + nxt_lvlhsh_alloc, + nxt_lvlhsh_free, +}; + + +static int +nxt_unit_request_hash_add(nxt_lvlhsh_t *request_hash, + nxt_unit_request_info_impl_t *req_impl) +{ + uint32_t *stream; + nxt_int_t res; + nxt_lvlhsh_query_t lhq; + + stream = &req_impl->stream; + + lhq.key_hash = nxt_murmur_hash2(stream, sizeof(*stream)); + lhq.key.length = sizeof(*stream); + lhq.key.start = (u_char *) stream; + lhq.proto = &lvlhsh_requests_proto; + lhq.pool = NULL; + lhq.replace = 0; + lhq.value = req_impl; + + res = nxt_lvlhsh_insert(request_hash, &lhq); + + switch (res) { + + case NXT_OK: + return NXT_UNIT_OK; + + default: + return NXT_UNIT_ERROR; + } +} + + +static nxt_unit_request_info_impl_t * +nxt_unit_request_hash_find(nxt_lvlhsh_t *request_hash, uint32_t stream, + int remove) +{ + nxt_int_t res; + nxt_lvlhsh_query_t lhq; + + lhq.key_hash = nxt_murmur_hash2(&stream, sizeof(stream)); + lhq.key.length = sizeof(stream); + lhq.key.start = (u_char *) &stream; + lhq.proto = &lvlhsh_requests_proto; + lhq.pool = NULL; + + if (remove) { + res = nxt_lvlhsh_delete(request_hash, &lhq); + + } else { + res = nxt_lvlhsh_find(request_hash, &lhq); + } + + switch (res) { + + case NXT_OK: + return lhq.value; + + default: + return NULL; + } +} + + void nxt_unit_log(nxt_unit_ctx_t *ctx, int level, const char *fmt, ...) { @@ -3526,8 +4144,7 @@ nxt_unit_req_log(nxt_unit_request_info_t *req, int level, const char *fmt, ...) if (nxt_fast_path(req != NULL)) { req_impl = nxt_container_of(req, nxt_unit_request_info_impl_t, req); - p += snprintf(p, end - p, - "#%"PRIu32": ", req_impl->recv_msg.port_msg.stream); + p += snprintf(p, end - p, "#%"PRIu32": ", req_impl->stream); } va_start(ap, fmt); |