nxt_unit.c (1398:05063d6eec8e) nxt_unit.c (1403:1cee885b7f10)
1
2/*
3 * Copyright (C) NGINX, Inc.
4 */
5
6#include <stdlib.h>
7
8#include "nxt_main.h"

--- 62 unchanged lines hidden (view full) ---

71 nxt_unit_mmap_buf_t *mmap_buf, int last);
72static void nxt_unit_mmap_buf_free(nxt_unit_mmap_buf_t *mmap_buf);
73static void nxt_unit_free_outgoing_buf(nxt_unit_mmap_buf_t *mmap_buf);
74static nxt_unit_read_buf_t *nxt_unit_read_buf_get(nxt_unit_ctx_t *ctx);
75static nxt_unit_read_buf_t *nxt_unit_read_buf_get_impl(
76 nxt_unit_ctx_impl_t *ctx_impl);
77static void nxt_unit_read_buf_release(nxt_unit_ctx_t *ctx,
78 nxt_unit_read_buf_t *rbuf);
1
2/*
3 * Copyright (C) NGINX, Inc.
4 */
5
6#include <stdlib.h>
7
8#include "nxt_main.h"

--- 62 unchanged lines hidden (view full) ---

71 nxt_unit_mmap_buf_t *mmap_buf, int last);
72static void nxt_unit_mmap_buf_free(nxt_unit_mmap_buf_t *mmap_buf);
73static void nxt_unit_free_outgoing_buf(nxt_unit_mmap_buf_t *mmap_buf);
74static nxt_unit_read_buf_t *nxt_unit_read_buf_get(nxt_unit_ctx_t *ctx);
75static nxt_unit_read_buf_t *nxt_unit_read_buf_get_impl(
76 nxt_unit_ctx_impl_t *ctx_impl);
77static void nxt_unit_read_buf_release(nxt_unit_ctx_t *ctx,
78 nxt_unit_read_buf_t *rbuf);
79static nxt_unit_mmap_buf_t *nxt_unit_request_preread(
80 nxt_unit_request_info_t *req, size_t size);
79static ssize_t nxt_unit_buf_read(nxt_unit_buf_t **b, uint64_t *len, void *dst,
80 size_t size);
81static nxt_port_mmap_header_t *nxt_unit_mmap_get(nxt_unit_ctx_t *ctx,
82 nxt_unit_process_t *process, nxt_unit_port_id_t *port_id,
83 nxt_chunk_id_t *c, int *n, int min_n);
84static int nxt_unit_send_oosm(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id);
85static int nxt_unit_wait_shm_ack(nxt_unit_ctx_t *ctx);
86static nxt_unit_mmap_t *nxt_unit_mmap_at(nxt_unit_mmaps_t *mmaps, uint32_t i);

--- 869 unchanged lines hidden (view full) ---

956 b->req = req;
957 }
958
959 /* "Move" incoming buffer list to req_impl. */
960 req_impl->incoming_buf = recv_msg->incoming_buf;
961 req_impl->incoming_buf->prev = &req_impl->incoming_buf;
962 recv_msg->incoming_buf = NULL;
963
81static ssize_t nxt_unit_buf_read(nxt_unit_buf_t **b, uint64_t *len, void *dst,
82 size_t size);
83static nxt_port_mmap_header_t *nxt_unit_mmap_get(nxt_unit_ctx_t *ctx,
84 nxt_unit_process_t *process, nxt_unit_port_id_t *port_id,
85 nxt_chunk_id_t *c, int *n, int min_n);
86static int nxt_unit_send_oosm(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id);
87static int nxt_unit_wait_shm_ack(nxt_unit_ctx_t *ctx);
88static nxt_unit_mmap_t *nxt_unit_mmap_at(nxt_unit_mmaps_t *mmaps, uint32_t i);

--- 869 unchanged lines hidden (view full) ---

958 b->req = req;
959 }
960
961 /* "Move" incoming buffer list to req_impl. */
962 req_impl->incoming_buf = recv_msg->incoming_buf;
963 req_impl->incoming_buf->prev = &req_impl->incoming_buf;
964 recv_msg->incoming_buf = NULL;
965
966 req->content_fd = recv_msg->fd;
967 recv_msg->fd = -1;
968
964 req->response_max_fields = 0;
965 req_impl->state = NXT_UNIT_RS_START;
966 req_impl->websocket = 0;
967
968 nxt_unit_debug(ctx, "#%"PRIu32": %.*s %.*s (%d)", recv_msg->stream,
969 (int) r->method_length, nxt_unit_sptr_get(&r->method),
970 (int) r->target_length, nxt_unit_sptr_get(&r->target),
971 (int) r->content_length);

--- 201 unchanged lines hidden (view full) ---

1173 while (req_impl->outgoing_buf != NULL) {
1174 nxt_unit_mmap_buf_free(req_impl->outgoing_buf);
1175 }
1176
1177 while (req_impl->incoming_buf != NULL) {
1178 nxt_unit_mmap_buf_free(req_impl->incoming_buf);
1179 }
1180
969 req->response_max_fields = 0;
970 req_impl->state = NXT_UNIT_RS_START;
971 req_impl->websocket = 0;
972
973 nxt_unit_debug(ctx, "#%"PRIu32": %.*s %.*s (%d)", recv_msg->stream,
974 (int) r->method_length, nxt_unit_sptr_get(&r->method),
975 (int) r->target_length, nxt_unit_sptr_get(&r->target),
976 (int) r->content_length);

--- 201 unchanged lines hidden (view full) ---

1178 while (req_impl->outgoing_buf != NULL) {
1179 nxt_unit_mmap_buf_free(req_impl->outgoing_buf);
1180 }
1181
1182 while (req_impl->incoming_buf != NULL) {
1183 nxt_unit_mmap_buf_free(req_impl->incoming_buf);
1184 }
1185
1186 if (req->content_fd != -1) {
1187 close(req->content_fd);
1188
1189 req->content_fd = -1;
1190 }
1191
1181 /*
1182 * Process release should go after buffers release to guarantee mmap
1183 * existence.
1184 */
1185 if (req_impl->process != NULL) {
1186 nxt_unit_process_use(req->ctx, req_impl->process, -1);
1187
1188 req_impl->process = NULL;

--- 1229 unchanged lines hidden (view full) ---

2418
2419 return NXT_UNIT_OK;
2420}
2421
2422
2423ssize_t
2424nxt_unit_request_read(nxt_unit_request_info_t *req, void *dst, size_t size)
2425{
1192 /*
1193 * Process release should go after buffers release to guarantee mmap
1194 * existence.
1195 */
1196 if (req_impl->process != NULL) {
1197 nxt_unit_process_use(req->ctx, req_impl->process, -1);
1198
1199 req_impl->process = NULL;

--- 1229 unchanged lines hidden (view full) ---

2429
2430 return NXT_UNIT_OK;
2431}
2432
2433
2434ssize_t
2435nxt_unit_request_read(nxt_unit_request_info_t *req, void *dst, size_t size)
2436{
2426 return nxt_unit_buf_read(&req->content_buf, &req->content_length,
2427 dst, size);
2437 ssize_t buf_res, res;
2438
2439 buf_res = nxt_unit_buf_read(&req->content_buf, &req->content_length,
2440 dst, size);
2441
2442 if (buf_res < (ssize_t) size && req->content_fd != -1) {
2443 res = read(req->content_fd, dst, size);
2444 if (res < 0) {
2445 nxt_unit_req_alert(req, "failed to read content: %s (%d)",
2446 strerror(errno), errno);
2447
2448 return res;
2449 }
2450
2451 if (res < (ssize_t) size) {
2452 close(req->content_fd);
2453
2454 req->content_fd = -1;
2455 }
2456
2457 req->content_length -= res;
2458 size -= res;
2459
2460 dst = nxt_pointer_to(dst, res);
2461
2462 } else {
2463 res = 0;
2464 }
2465
2466 return buf_res + res;
2428}
2429
2430
2431ssize_t
2432nxt_unit_request_readline_size(nxt_unit_request_info_t *req, size_t max_size)
2433{
2467}
2468
2469
2470ssize_t
2471nxt_unit_request_readline_size(nxt_unit_request_info_t *req, size_t max_size)
2472{
2434 char *p;
2435 size_t l_size, b_size;
2436 nxt_unit_buf_t *b;
2473 char *p;
2474 size_t l_size, b_size;
2475 nxt_unit_buf_t *b;
2476 nxt_unit_mmap_buf_t *mmap_buf, *preread_buf;
2437
2438 if (req->content_length == 0) {
2439 return 0;
2440 }
2441
2442 l_size = 0;
2443
2444 b = req->content_buf;

--- 9 unchanged lines hidden (view full) ---

2454 }
2455
2456 l_size += b_size;
2457
2458 if (max_size <= l_size) {
2459 break;
2460 }
2461
2477
2478 if (req->content_length == 0) {
2479 return 0;
2480 }
2481
2482 l_size = 0;
2483
2484 b = req->content_buf;

--- 9 unchanged lines hidden (view full) ---

2494 }
2495
2496 l_size += b_size;
2497
2498 if (max_size <= l_size) {
2499 break;
2500 }
2501
2502 mmap_buf = nxt_container_of(b, nxt_unit_mmap_buf_t, buf);
2503 if (mmap_buf->next == NULL
2504 && req->content_fd != -1
2505 && l_size < req->content_length)
2506 {
2507 preread_buf = nxt_unit_request_preread(req, 16384);
2508 if (nxt_slow_path(preread_buf == NULL)) {
2509 return -1;
2510 }
2511
2512 nxt_unit_mmap_buf_insert(&mmap_buf->next, preread_buf);
2513 }
2514
2462 b = nxt_unit_buf_next(b);
2463 }
2464
2465 return nxt_min(max_size, l_size);
2466}
2467
2468
2515 b = nxt_unit_buf_next(b);
2516 }
2517
2518 return nxt_min(max_size, l_size);
2519}
2520
2521
2522static nxt_unit_mmap_buf_t *
2523nxt_unit_request_preread(nxt_unit_request_info_t *req, size_t size)
2524{
2525 ssize_t res;
2526 nxt_unit_mmap_buf_t *mmap_buf;
2527
2528 if (req->content_fd == -1) {
2529 nxt_unit_req_alert(req, "preread: content_fd == -1");
2530 return NULL;
2531 }
2532
2533 mmap_buf = nxt_unit_mmap_buf_get(req->ctx);
2534 if (nxt_slow_path(mmap_buf == NULL)) {
2535 nxt_unit_req_alert(req, "preread: failed to allocate buf");
2536 return NULL;
2537 }
2538
2539 mmap_buf->free_ptr = malloc(size);
2540 if (nxt_slow_path(mmap_buf->free_ptr == NULL)) {
2541 nxt_unit_req_alert(req, "preread: failed to allocate buf memory");
2542 nxt_unit_mmap_buf_release(mmap_buf);
2543 return NULL;
2544 }
2545
2546 mmap_buf->plain_ptr = mmap_buf->free_ptr;
2547
2548 mmap_buf->hdr = NULL;
2549 mmap_buf->buf.start = mmap_buf->free_ptr;
2550 mmap_buf->buf.free = mmap_buf->buf.start;
2551 mmap_buf->buf.end = mmap_buf->buf.start + size;
2552 mmap_buf->process = NULL;
2553
2554 res = read(req->content_fd, mmap_buf->free_ptr, size);
2555 if (res < 0) {
2556 nxt_unit_req_alert(req, "failed to read content: %s (%d)",
2557 strerror(errno), errno);
2558
2559 nxt_unit_mmap_buf_free(mmap_buf);
2560
2561 return NULL;
2562 }
2563
2564 if (res < (ssize_t) size) {
2565 close(req->content_fd);
2566
2567 req->content_fd = -1;
2568 }
2569
2570 nxt_unit_req_debug(req, "preread: read %d", (int) res);
2571
2572 mmap_buf->buf.end = mmap_buf->buf.free + res;
2573
2574 return mmap_buf;
2575}
2576
2577
2469static ssize_t
2470nxt_unit_buf_read(nxt_unit_buf_t **b, uint64_t *len, void *dst, size_t size)
2471{
2472 u_char *p;
2473 size_t rest, copy, read;
2578static ssize_t
2579nxt_unit_buf_read(nxt_unit_buf_t **b, uint64_t *len, void *dst, size_t size)
2580{
2581 u_char *p;
2582 size_t rest, copy, read;
2474 nxt_unit_buf_t *buf;
2583 nxt_unit_buf_t *buf, *last_buf;
2475
2476 p = dst;
2477 rest = size;
2478
2479 buf = *b;
2584
2585 p = dst;
2586 rest = size;
2587
2588 buf = *b;
2589 last_buf = buf;
2480
2481 while (buf != NULL) {
2590
2591 while (buf != NULL) {
2592 last_buf = buf;
2593
2482 copy = buf->end - buf->free;
2483 copy = nxt_min(rest, copy);
2484
2485 p = nxt_cpymem(p, buf->free, copy);
2486
2487 buf->free += copy;
2488 rest -= copy;
2489
2490 if (rest == 0) {
2491 if (buf->end == buf->free) {
2492 buf = nxt_unit_buf_next(buf);
2493 }
2494
2495 break;
2496 }
2497
2498 buf = nxt_unit_buf_next(buf);
2499 }
2500
2594 copy = buf->end - buf->free;
2595 copy = nxt_min(rest, copy);
2596
2597 p = nxt_cpymem(p, buf->free, copy);
2598
2599 buf->free += copy;
2600 rest -= copy;
2601
2602 if (rest == 0) {
2603 if (buf->end == buf->free) {
2604 buf = nxt_unit_buf_next(buf);
2605 }
2606
2607 break;
2608 }
2609
2610 buf = nxt_unit_buf_next(buf);
2611 }
2612
2501 *b = buf;
2613 *b = last_buf;
2502
2503 read = size - rest;
2504
2505 *len -= read;
2506
2507 return read;
2508}
2509

--- 2355 unchanged lines hidden --- 		2614
2615 read = size - rest;
2616
2617 *len -= read;
2618
2619 return read;
2620}
2621

--- 2355 unchanged lines hidden ---