One Level Up Top Level

src/event/ngx_event_pipe.c - nginx-1.31.7 nginx/ @ 939334eff

Functions defined

Source code


  2. /*
  3. * Copyright (C) Igor Sysoev
  4. * Copyright (C) Nginx, Inc.
  5. */


  8. #include <ngx_config.h>
  9. #include <ngx_core.h>
  10. #include <ngx_event.h>
  11. #include <ngx_event_pipe.h>


  14. static ngx_int_t ngx_event_pipe_read_upstream(ngx_event_pipe_t *p);
  15. static ngx_int_t ngx_event_pipe_write_to_downstream(ngx_event_pipe_t *p);

  17. static ngx_int_t ngx_event_pipe_write_chain_to_temp_file(ngx_event_pipe_t *p);
  18. static ngx_inline void ngx_event_pipe_remove_shadow_links(ngx_buf_t *buf);
  19. static ngx_int_t ngx_event_pipe_drain_chains(ngx_event_pipe_t *p);


  22. ngx_int_t
  23. ‌ngx_event_pipe(ngx_event_pipe_t *p, ngx_int_t do_write)
  24. {
  25.     ngx_int_t     rc;
  26.     ngx_uint_t    flags;
  27.     ngx_event_t  *rev, *wev;

  29.     for ( ;; ) {
  30.         if (do_write) {
  31.             p->log->action = "sending to client";

  33.             rc = ngx_event_pipe_write_to_downstream(p);

  35.             if (rc == NGX_ABORT) {
  36.                 return NGX_ABORT;
  37.             }

  39.             if (rc == NGX_BUSY) {
  40.                 return NGX_OK;
  41.             }
  42.         }

  44.         p->read = 0;
  45.         p->upstream_blocked = 0;

  47.         p->log->action = "reading upstream";

  49.         if (ngx_event_pipe_read_upstream(p) == NGX_ABORT) {
  50.             return NGX_ABORT;
  51.         }

  53.         if (!p->read && !p->upstream_blocked) {
  54.             break;
  55.         }

  57.         do_write = 1;
  58.     }

  60.     if (p->upstream
  61.         && p->upstream->fd != (ngx_socket_t) -1)
  62.     {
  63.         rev = p->upstream->read;

  65.         flags = (rev->eof || rev->error) ? NGX_CLOSE_EVENT : 0;

  67.         if (ngx_handle_read_event(rev, flags) != NGX_OK) {
  68.             return NGX_ABORT;
  69.         }

  71.         if (!rev->delayed) {
  72.             if (rev->active && !rev->ready) {
  73.                 ngx_add_timer(rev, p->read_timeout);

  75.             } else if (rev->timer_set) {
  76.                 ngx_del_timer(rev);
  77.             }
  78.         }
  79.     }

  81.     if (p->downstream->fd != (ngx_socket_t) -1
  82.         && p->downstream->data == p->output_ctx)
  83.     {
  84.         wev = p->downstream->write;
  85.         if (ngx_handle_write_event(wev, p->send_lowat) != NGX_OK) {
  86.             return NGX_ABORT;
  87.         }

  89.         if (!wev->delayed) {
  90.             if (wev->active && !wev->ready) {
  91.                 ngx_add_timer(wev, p->send_timeout);

  93.             } else if (wev->timer_set) {
  94.                 ngx_del_timer(wev);
  95.             }
  96.         }
  97.     }

  99.     return NGX_OK;
  100. }


  103. static ngx_int_t
  104. ‌ngx_event_pipe_read_upstream(ngx_event_pipe_t *p)
  105. {
  106.     off_t         limit;
  107.     ssize_t       n, size;
  108.     ngx_int_t     rc;
  109.     ngx_buf_t    *b;
  110.     ngx_msec_t    delay;
  111.     ngx_chain_t  *chain, *cl, *ln;

  113.     if (p->upstream_eof || p->upstream_error || p->upstream_done
  114.         || p->upstream == NULL)
  115.     {
  116.         return NGX_OK;
  117.     }

  119. #if (NGX_THREADS)

  121.     if (p->aio) {
  122.         ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  123.                        "pipe read upstream: aio");
  124.         return NGX_AGAIN;
  125.     }

  127.     if (p->writing) {
  128.         ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  129.                        "pipe read upstream: writing");

  131.         rc = ngx_event_pipe_write_chain_to_temp_file(p);

  133.         if (rc != NGX_OK) {
  134.             return rc;
  135.         }
  136.     }

  138. #endif

  140.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
  141.                    "pipe read upstream: %d", p->upstream->read->ready);

  143.     for ( ;; ) {

  145.         if (p->upstream_eof || p->upstream_error || p->upstream_done) {
  146.             break;
  147.         }

  149.         if (p->preread_bufs == NULL && !p->upstream->read->ready) {
  150.             break;
  151.         }

  153.         if (p->preread_bufs) {

  155.             /* use the pre-read bufs if they exist */

  157.             chain = p->preread_bufs;
  158.             p->preread_bufs = NULL;
  159.             n = p->preread_size;

  161.             ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
  162.                            "pipe preread: %z", n);

  164.             if (n) {
  165.                 p->read = 1;
  166.             }

  168.         } else {

  170. #if (NGX_HAVE_KQUEUE)

  172.             /*
  173.              * kqueue notifies about the end of file or a pending error.
  174.              * This test allows not to allocate a buf on these conditions
  175.              * and not to call c->recv_chain().
  176.              */

  178.             if (p->upstream->read->available == 0
  179.                 && p->upstream->read->pending_eof
  180. #if (NGX_SSL)
  181.                 && !p->upstream->ssl
  182. #endif
  183.                 )
  184.             {
  185.                 p->upstream->read->ready = 0;
  186.                 p->upstream->read->eof = 1;
  187.                 p->upstream_eof = 1;
  188.                 p->read = 1;

  190.                 if (p->upstream->read->kq_errno) {
  191.                     p->upstream->read->error = 1;
  192.                     p->upstream_error = 1;
  193.                     p->upstream_eof = 0;

  195.                     ngx_log_error(NGX_LOG_ERR, p->log,
  196.                                   p->upstream->read->kq_errno,
  197.                                   "kevent() reported that upstream "
  198.                                   "closed connection");
  199.                 }

  201.                 break;
  202.             }
  203. #endif

  205.             if (p->limit_rate) {
  206.                 if (p->upstream->read->delayed) {
  207.                     break;
  208.                 }

  210.                 limit = (off_t) p->limit_rate * (ngx_time() - p->start_sec + 1)
  211.                         - p->read_length;

  213.                 if (limit <= 0) {
  214.                     p->upstream->read->delayed = 1;
  215.                     delay = (ngx_msec_t) (- limit * 1000 / p->limit_rate + 1);
  216.                     ngx_add_timer(p->upstream->read, delay);
  217.                     break;
  218.                 }

  220.             } else {
  221.                 limit = 0;
  222.             }

  224.             if (p->free_raw_bufs) {

  226.                 /* use the free bufs if they exist */

  228.                 chain = p->free_raw_bufs;
  229.                 if (p->single_buf) {
  230.                     p->free_raw_bufs = p->free_raw_bufs->next;
  231.                     chain->next = NULL;
  232.                 } else {
  233.                     p->free_raw_bufs = NULL;
  234.                 }

  236.             } else if (p->allocated < p->bufs.num) {

  238.                 /* allocate a new buf if it's still allowed */

  240.                 b = ngx_create_temp_buf(p->pool, p->bufs.size);
  241.                 if (b == NULL) {
  242.                     return NGX_ABORT;
  243.                 }

  245.                 p->allocated++;

  247.                 chain = ngx_alloc_chain_link(p->pool);
  248.                 if (chain == NULL) {
  249.                     return NGX_ABORT;
  250.                 }

  252.                 chain->buf = b;
  253.                 chain->next = NULL;

  255.             } else if (!p->cacheable
  256.                        && !p->downstream_error
  257.                        && p->downstream->data == p->output_ctx
  258.                        && p->downstream->write->ready
  259.                        && !p->downstream->write->delayed)
  260.             {
  261.                 /*
  262.                  * if the bufs are not needed to be saved in a cache and
  263.                  * a downstream is ready then write the bufs to a downstream
  264.                  */

  266.                 p->upstream_blocked = 1;

  268.                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  269.                                "pipe downstream ready");

  271.                 break;

  273.             } else if (p->cacheable
  274.                        || p->temp_file->offset < p->max_temp_file_size)
  275.             {

  277.                 /*
  278.                  * if it is allowed, then save some bufs from p->in
  279.                  * to a temporary file, and add them to a p->out chain
  280.                  */

  282.                 rc = ngx_event_pipe_write_chain_to_temp_file(p);

  284.                 ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
  285.                                "pipe temp offset: %O", p->temp_file->offset);

  287.                 if (rc == NGX_BUSY) {
  288.                     break;
  289.                 }

  291.                 if (rc != NGX_OK) {
  292.                     return rc;
  293.                 }

  295.                 chain = p->free_raw_bufs;
  296.                 if (p->single_buf) {
  297.                     p->free_raw_bufs = p->free_raw_bufs->next;
  298.                     chain->next = NULL;
  299.                 } else {
  300.                     p->free_raw_bufs = NULL;
  301.                 }

  303.             } else {

  305.                 /* there are no bufs to read in */

  307.                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  308.                                "no pipe bufs to read in");

  310.                 break;
  311.             }

  313.             n = p->upstream->recv_chain(p->upstream, chain, limit);

  315.             ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
  316.                            "pipe recv chain: %z", n);

  318.             if (p->free_raw_bufs) {
  319.                 chain->next = p->free_raw_bufs;
  320.             }
  321.             p->free_raw_bufs = chain;

  323.             if (n == NGX_ERROR) {
  324.                 p->upstream_error = 1;
  325.                 break;
  326.             }

  328.             if (n == NGX_AGAIN) {
  329.                 if (p->single_buf) {
  330.                     ngx_event_pipe_remove_shadow_links(chain->buf);
  331.                 }

  333.                 break;
  334.             }

  336.             p->read = 1;

  338.             if (n == 0) {
  339.                 p->upstream_eof = 1;
  340.                 break;
  341.             }
  342.         }

  344.         delay = p->limit_rate ? (ngx_msec_t) n * 1000 / p->limit_rate : 0;

  346.         p->read_length += n;
  347.         cl = chain;
  348.         p->free_raw_bufs = NULL;

  350.         while (cl && n > 0) {

  352.             ngx_event_pipe_remove_shadow_links(cl->buf);

  354.             size = cl->buf->end - cl->buf->last;

  356.             if (n >= size) {
  357.                 cl->buf->last = cl->buf->end;

  359.                 /* STUB */ cl->buf->num = p->num++;

  361.                 if (p->input_filter(p, cl->buf) == NGX_ERROR) {
  362.                     return NGX_ABORT;
  363.                 }

  365.                 n -= size;
  366.                 ln = cl;
  367.                 cl = cl->next;
  368.                 ngx_free_chain(p->pool, ln);

  370.             } else {
  371.                 cl->buf->last += n;
  372.                 n = 0;
  373.             }
  374.         }

  376.         if (cl) {
  377.             for (ln = cl; ln->next; ln = ln->next) { /* void */ }

  379.             ln->next = p->free_raw_bufs;
  380.             p->free_raw_bufs = cl;
  381.         }

  383.         if (delay > 0) {
  384.             p->upstream->read->delayed = 1;
  385.             ngx_add_timer(p->upstream->read, delay);
  386.             break;
  387.         }
  388.     }

  390. #if (NGX_DEBUG)

  392.     for (cl = p->busy; cl; cl = cl->next) {
  393.         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
  394.                        "pipe buf busy s:%d t:%d f:%d "
  395.                        "%p, pos %p, size: %z "
  396.                        "file: %O, size: %O",
  397.                        (cl->buf->shadow ? 1 : 0),
  398.                        cl->buf->temporary, cl->buf->in_file,
  399.                        cl->buf->start, cl->buf->pos,
  400.                        cl->buf->last - cl->buf->pos,
  401.                        cl->buf->file_pos,
  402.                        cl->buf->file_last - cl->buf->file_pos);
  403.     }

  405.     for (cl = p->out; cl; cl = cl->next) {
  406.         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
  407.                        "pipe buf out  s:%d t:%d f:%d "
  408.                        "%p, pos %p, size: %z "
  409.                        "file: %O, size: %O",
  410.                        (cl->buf->shadow ? 1 : 0),
  411.                        cl->buf->temporary, cl->buf->in_file,
  412.                        cl->buf->start, cl->buf->pos,
  413.                        cl->buf->last - cl->buf->pos,
  414.                        cl->buf->file_pos,
  415.                        cl->buf->file_last - cl->buf->file_pos);
  416.     }

  418.     for (cl = p->in; cl; cl = cl->next) {
  419.         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
  420.                        "pipe buf in   s:%d t:%d f:%d "
  421.                        "%p, pos %p, size: %z "
  422.                        "file: %O, size: %O",
  423.                        (cl->buf->shadow ? 1 : 0),
  424.                        cl->buf->temporary, cl->buf->in_file,
  425.                        cl->buf->start, cl->buf->pos,
  426.                        cl->buf->last - cl->buf->pos,
  427.                        cl->buf->file_pos,
  428.                        cl->buf->file_last - cl->buf->file_pos);
  429.     }

  431.     for (cl = p->free_raw_bufs; cl; cl = cl->next) {
  432.         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
  433.                        "pipe buf free s:%d t:%d f:%d "
  434.                        "%p, pos %p, size: %z "
  435.                        "file: %O, size: %O",
  436.                        (cl->buf->shadow ? 1 : 0),
  437.                        cl->buf->temporary, cl->buf->in_file,
  438.                        cl->buf->start, cl->buf->pos,
  439.                        cl->buf->last - cl->buf->pos,
  440.                        cl->buf->file_pos,
  441.                        cl->buf->file_last - cl->buf->file_pos);
  442.     }

  444.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
  445.                    "pipe length: %O", p->length);

  447. #endif

  449.     if (p->free_raw_bufs && p->length != -1) {
  450.         cl = p->free_raw_bufs;

  452.         if (cl->buf->last - cl->buf->pos >= p->length) {

  454.             p->free_raw_bufs = cl->next;

  456.             /* STUB */ cl->buf->num = p->num++;

  458.             if (p->input_filter(p, cl->buf) == NGX_ERROR) {
  459.                 return NGX_ABORT;
  460.             }

  462.             ngx_free_chain(p->pool, cl);
  463.         }
  464.     }

  466.     if (p->length == 0) {
  467.         p->upstream_done = 1;
  468.         p->read = 1;
  469.     }

  471.     if ((p->upstream_eof || p->upstream_error) && p->free_raw_bufs) {

  473.         /* STUB */ p->free_raw_bufs->buf->num = p->num++;

  475.         if (p->input_filter(p, p->free_raw_bufs->buf) == NGX_ERROR) {
  476.             return NGX_ABORT;
  477.         }

  479.         p->free_raw_bufs = p->free_raw_bufs->next;

  481.         if (p->free_bufs && p->buf_to_file == NULL) {
  482.             for (cl = p->free_raw_bufs; cl; cl = cl->next) {
  483.                 if (cl->buf->shadow == NULL) {
  484.                     ngx_pfree(p->pool, cl->buf->start);
  485.                 }
  486.             }
  487.         }
  488.     }

  490.     if (p->cacheable && (p->in || p->buf_to_file)) {

  492.         ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  493.                        "pipe write chain");

  495.         rc = ngx_event_pipe_write_chain_to_temp_file(p);

  497.         if (rc != NGX_OK) {
  498.             return rc;
  499.         }
  500.     }

  502.     return NGX_OK;
  503. }


  506. static ngx_int_t
  507. ‌ngx_event_pipe_write_to_downstream(ngx_event_pipe_t *p)
  508. {
  509.     u_char            *prev;
  510.     size_t             bsize;
  511.     ngx_int_t          rc;
  512.     ngx_uint_t         flush, flushed, prev_last_shadow;
  513.     ngx_chain_t       *out, **ll, *cl;
  514.     ngx_connection_t  *downstream;

  516.     downstream = p->downstream;

  518.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
  519.                    "pipe write downstream: %d", downstream->write->ready);

  521. #if (NGX_THREADS)

  523.     if (p->writing) {
  524.         rc = ngx_event_pipe_write_chain_to_temp_file(p);

  526.         if (rc == NGX_ABORT) {
  527.             return NGX_ABORT;
  528.         }
  529.     }

  531. #endif

  533.     flushed = 0;

  535.     for ( ;; ) {
  536.         if (p->downstream_error) {
  537.             return ngx_event_pipe_drain_chains(p);
  538.         }

  540.         if (p->upstream_eof || p->upstream_error || p->upstream_done) {

  542.             /* pass the p->out and p->in chains to the output filter */

  544.             for (cl = p->busy; cl; cl = cl->next) {
  545.                 cl->buf->recycled = 0;
  546.             }

  548.             if (p->out) {
  549.                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  550.                                "pipe write downstream flush out");

  552.                 for (cl = p->out; cl; cl = cl->next) {
  553.                     cl->buf->recycled = 0;
  554.                 }

  556.                 rc = p->output_filter(p->output_ctx, p->out);

  558.                 if (rc == NGX_ERROR) {
  559.                     p->downstream_error = 1;
  560.                     return ngx_event_pipe_drain_chains(p);
  561.                 }

  563.                 p->out = NULL;
  564.             }

  566.             if (p->writing) {
  567.                 break;
  568.             }

  570.             if (p->in) {
  571.                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  572.                                "pipe write downstream flush in");

  574.                 for (cl = p->in; cl; cl = cl->next) {
  575.                     cl->buf->recycled = 0;
  576.                 }

  578.                 rc = p->output_filter(p->output_ctx, p->in);

  580.                 if (rc == NGX_ERROR) {
  581.                     p->downstream_error = 1;
  582.                     return ngx_event_pipe_drain_chains(p);
  583.                 }

  585.                 p->in = NULL;
  586.             }

  588.             ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  589.                            "pipe write downstream done");

  591.             /* TODO: free unused bufs */

  593.             p->downstream_done = 1;
  594.             break;
  595.         }

  597.         if (downstream->data != p->output_ctx
  598.             || !downstream->write->ready
  599.             || downstream->write->delayed)
  600.         {
  601.             break;
  602.         }

  604.         /* bsize is the size of the busy recycled bufs */

  606.         prev = NULL;
  607.         bsize = 0;

  609.         for (cl = p->busy; cl; cl = cl->next) {

  611.             if (cl->buf->recycled) {
  612.                 if (prev == cl->buf->start) {
  613.                     continue;
  614.                 }

  616.                 bsize += cl->buf->end - cl->buf->start;
  617.                 prev = cl->buf->start;
  618.             }
  619.         }

  621.         ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
  622.                        "pipe write busy: %uz", bsize);

  624.         out = NULL;

  626.         if (bsize >= (size_t) p->busy_size) {
  627.             flush = 1;
  628.             goto flush;
  629.         }

  631.         flush = 0;
  632.         ll = NULL;
  633.         prev_last_shadow = 1;

  635.         for ( ;; ) {
  636.             if (p->out) {
  637.                 cl = p->out;

  639.                 if (cl->buf->recycled) {
  640.                     ngx_log_error(NGX_LOG_ALERT, p->log, 0,
  641.                                   "recycled buffer in pipe out chain");
  642.                 }

  644.                 p->out = p->out->next;

  646.             } else if (!p->cacheable && !p->writing && p->in) {
  647.                 cl = p->in;

  649.                 ngx_log_debug3(NGX_LOG_DEBUG_EVENT, p->log, 0,
  650.                                "pipe write buf ls:%d %p %z",
  651.                                cl->buf->last_shadow,
  652.                                cl->buf->pos,
  653.                                cl->buf->last - cl->buf->pos);

  655.                 if (cl->buf->recycled && prev_last_shadow) {
  656.                     if (bsize + cl->buf->end - cl->buf->start > p->busy_size) {
  657.                         flush = 1;
  658.                         break;
  659.                     }

  661.                     bsize += cl->buf->end - cl->buf->start;
  662.                 }

  664.                 prev_last_shadow = cl->buf->last_shadow;

  666.                 p->in = p->in->next;

  668.             } else {
  669.                 break;
  670.             }

  672.             cl->next = NULL;

  674.             if (out) {
  675.                 *ll = cl;
  676.             } else {
  677.                 out = cl;
  678.             }
  679.             ll = &cl->next;
  680.         }

  682.     flush:

  684.         ngx_log_debug2(NGX_LOG_DEBUG_EVENT, p->log, 0,
  685.                        "pipe write: out:%p, f:%ui", out, flush);

  687.         if (out == NULL) {

  689.             if (!flush) {
  690.                 break;
  691.             }

  693.             /* a workaround for AIO */
  694.             if (flushed++ > 10) {
  695.                 return NGX_BUSY;
  696.             }
  697.         }

  699.         rc = p->output_filter(p->output_ctx, out);

  701.         ngx_chain_update_chains(p->pool, &p->free, &p->busy, &out, p->tag);

  703.         if (rc == NGX_ERROR) {
  704.             p->downstream_error = 1;
  705.             return ngx_event_pipe_drain_chains(p);
  706.         }

  708.         for (cl = p->free; cl; cl = cl->next) {

  710.             if (cl->buf->temp_file) {
  711.                 if (p->cacheable || !p->cyclic_temp_file) {
  712.                     continue;
  713.                 }

  715.                 /* reset p->temp_offset if all bufs had been sent */

  717.                 if (cl->buf->file_last == p->temp_file->offset) {
  718.                     p->temp_file->offset = 0;
  719.                 }
  720.             }

  722.             /* TODO: free buf if p->free_bufs && upstream done */

  724.             /* add the free shadow raw buf to p->free_raw_bufs */

  726.             if (cl->buf->last_shadow) {
  727.                 if (ngx_event_pipe_add_free_buf(p, cl->buf->shadow) != NGX_OK) {
  728.                     return NGX_ABORT;
  729.                 }

  731.                 cl->buf->last_shadow = 0;
  732.             }

  734.             cl->buf->shadow = NULL;
  735.         }
  736.     }

  738.     return NGX_OK;
  739. }


  742. static ngx_int_t
  743. ‌ngx_event_pipe_write_chain_to_temp_file(ngx_event_pipe_t *p)
  744. {
  745.     ssize_t       size, bsize, n;
  746.     ngx_buf_t    *b;
  747.     ngx_uint_t    prev_last_shadow;
  748.     ngx_chain_t  *cl, *tl, *next, *out, **ll, **last_out, **last_free;

  750. #if (NGX_THREADS)

  752.     if (p->writing) {

  754.         if (p->aio) {
  755.             return NGX_AGAIN;
  756.         }

  758.         out = p->writing;
  759.         p->writing = NULL;

  761.         n = ngx_write_chain_to_temp_file(p->temp_file, NULL);

  763.         if (n == NGX_ERROR) {
  764.             return NGX_ABORT;
  765.         }

  767.         goto done;
  768.     }

  770. #endif

  772.     if (p->buf_to_file) {
  773.         out = ngx_alloc_chain_link(p->pool);
  774.         if (out == NULL) {
  775.             return NGX_ABORT;
  776.         }

  778.         out->buf = p->buf_to_file;
  779.         out->next = p->in;

  781.     } else {
  782.         out = p->in;
  783.     }

  785.     if (!p->cacheable) {

  787.         size = 0;
  788.         cl = out;
  789.         ll = NULL;
  790.         prev_last_shadow = 1;

  792.         ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
  793.                        "pipe offset: %O", p->temp_file->offset);

  795.         do {
  796.             bsize = cl->buf->last - cl->buf->pos;

  798.             ngx_log_debug4(NGX_LOG_DEBUG_EVENT, p->log, 0,
  799.                            "pipe buf ls:%d %p, pos %p, size: %z",
  800.                            cl->buf->last_shadow, cl->buf->start,
  801.                            cl->buf->pos, bsize);

  803.             if (prev_last_shadow
  804.                 && ((size + bsize > p->temp_file_write_size)
  805.                     || (p->temp_file->offset + size + bsize
  806.                         > p->max_temp_file_size)))
  807.             {
  808.                 break;
  809.             }

  811.             prev_last_shadow = cl->buf->last_shadow;

  813.             size += bsize;
  814.             ll = &cl->next;
  815.             cl = cl->next;

  817.         } while (cl);

  819.         ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0, "size: %z", size);

  821.         if (ll == NULL) {
  822.             return NGX_BUSY;
  823.         }

  825.         if (cl) {
  826.             p->in = cl;
  827.             *ll = NULL;

  829.         } else {
  830.             p->in = NULL;
  831.             p->last_in = &p->in;
  832.         }

  834.     } else {
  835.         p->in = NULL;
  836.         p->last_in = &p->in;
  837.     }

  839. #if (NGX_THREADS)
  840.     if (p->thread_handler) {
  841.         p->temp_file->thread_write = 1;
  842.         p->temp_file->file.thread_task = p->thread_task;
  843.         p->temp_file->file.thread_handler = p->thread_handler;
  844.         p->temp_file->file.thread_ctx = p->thread_ctx;
  845.     }
  846. #endif

  848.     n = ngx_write_chain_to_temp_file(p->temp_file, out);

  850.     if (n == NGX_ERROR) {
  851.         return NGX_ABORT;
  852.     }

  854. #if (NGX_THREADS)

  856.     if (n == NGX_AGAIN) {
  857.         p->writing = out;
  858.         p->thread_task = p->temp_file->file.thread_task;
  859.         return NGX_AGAIN;
  860.     }

  862. done:

  864. #endif

  866.     if (p->buf_to_file) {
  867.         p->temp_file->offset = p->buf_to_file->last - p->buf_to_file->pos;
  868.         n -= p->buf_to_file->last - p->buf_to_file->pos;
  869.         p->buf_to_file = NULL;
  870.         out = out->next;
  871.     }

  873.     if (n > 0) {
  874.         /* update previous buffer or add new buffer */

  876.         if (p->out) {
  877.             for (cl = p->out; cl->next; cl = cl->next) { /* void */ }

  879.             b = cl->buf;

  881.             if (b->file_last == p->temp_file->offset) {
  882.                 p->temp_file->offset += n;
  883.                 b->file_last = p->temp_file->offset;
  884.                 goto free;
  885.             }

  887.             last_out = &cl->next;

  889.         } else {
  890.             last_out = &p->out;
  891.         }

  893.         cl = ngx_chain_get_free_buf(p->pool, &p->free);
  894.         if (cl == NULL) {
  895.             return NGX_ABORT;
  896.         }

  898.         b = cl->buf;

  900.         ngx_memzero(b, sizeof(ngx_buf_t));

  902.         b->tag = p->tag;

  904.         b->file = &p->temp_file->file;
  905.         b->file_pos = p->temp_file->offset;
  906.         p->temp_file->offset += n;
  907.         b->file_last = p->temp_file->offset;

  909.         b->in_file = 1;
  910.         b->temp_file = 1;

  912.         *last_out = cl;
  913.     }

  915. free:

  917.     for (last_free = &p->free_raw_bufs;
  918.          *last_free != NULL;
  919.          last_free = &(*last_free)->next)
  920.     {
  921.         /* void */
  922.     }

  924.     for (cl = out; cl; cl = next) {
  925.         next = cl->next;

  927.         cl->next = p->free;
  928.         p->free = cl;

  930.         b = cl->buf;

  932.         if (b->last_shadow) {

  934.             tl = ngx_alloc_chain_link(p->pool);
  935.             if (tl == NULL) {
  936.                 return NGX_ABORT;
  937.             }

  939.             tl->buf = b->shadow;
  940.             tl->next = NULL;

  942.             *last_free = tl;
  943.             last_free = &tl->next;

  945.             b->shadow->pos = b->shadow->start;
  946.             b->shadow->last = b->shadow->start;

  948.             ngx_event_pipe_remove_shadow_links(b->shadow);
  949.         }
  950.     }

  952.     return NGX_OK;
  953. }


  956. /* the copy input filter */

  958. ngx_int_t
  959. ‌ngx_event_pipe_copy_input_filter(ngx_event_pipe_t *p, ngx_buf_t *buf)
  960. {
  961.     ngx_buf_t    *b;
  962.     ngx_chain_t  *cl;

  964.     if (buf->pos == buf->last) {
  965.         return NGX_OK;
  966.     }

  968.     if (p->upstream_done) {
  969.         ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  970.                        "input data after close");
  971.         return NGX_OK;
  972.     }

  974.     if (p->length == 0) {
  975.         p->upstream_done = 1;

  977.         ngx_log_error(NGX_LOG_WARN, p->log, 0,
  978.                       "upstream sent more data than specified in "
  979.                       "\"Content-Length\" header");

  981.         return NGX_OK;
  982.     }

  984.     cl = ngx_chain_get_free_buf(p->pool, &p->free);
  985.     if (cl == NULL) {
  986.         return NGX_ERROR;
  987.     }

  989.     b = cl->buf;

  991.     ngx_memcpy(b, buf, sizeof(ngx_buf_t));
  992.     b->shadow = buf;
  993.     b->tag = p->tag;
  994.     b->last_shadow = 1;
  995.     b->recycled = 1;
  996.     buf->shadow = b;

  998.     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0, "input buf #%d", b->num);

  1000.     if (p->in) {
  1001.         *p->last_in = cl;
  1002.     } else {
  1003.         p->in = cl;
  1004.     }
  1005.     p->last_in = &cl->next;

  1007.     if (p->length == -1) {
  1008.         return NGX_OK;
  1009.     }

  1011.     if (b->last - b->pos > p->length) {

  1013.         ngx_log_error(NGX_LOG_WARN, p->log, 0,
  1014.                       "upstream sent more data than specified in "
  1015.                       "\"Content-Length\" header");

  1017.         b->last = b->pos + p->length;
  1018.         p->upstream_done = 1;

  1020.         return NGX_OK;
  1021.     }

  1023.     p->length -= b->last - b->pos;

  1025.     return NGX_OK;
  1026. }


  1029. static ngx_inline void
  1030. ‌ngx_event_pipe_remove_shadow_links(ngx_buf_t *buf)
  1031. {
  1032.     ngx_buf_t  *b, *next;

  1034.     b = buf->shadow;

  1036.     if (b == NULL) {
  1037.         return;
  1038.     }

  1040.     while (!b->last_shadow) {
  1041.         next = b->shadow;

  1043.         b->temporary = 0;
  1044.         b->recycled = 0;

  1046.         b->shadow = NULL;
  1047.         b = next;
  1048.     }

  1050.     b->temporary = 0;
  1051.     b->recycled = 0;
  1052.     b->last_shadow = 0;

  1054.     b->shadow = NULL;

  1056.     buf->shadow = NULL;
  1057. }


  1060. ngx_int_t
  1061. ‌ngx_event_pipe_add_free_buf(ngx_event_pipe_t *p, ngx_buf_t *b)
  1062. {
  1063.     ngx_chain_t  *cl;

  1065.     cl = ngx_alloc_chain_link(p->pool);
  1066.     if (cl == NULL) {
  1067.         return NGX_ERROR;
  1068.     }

  1070.     if (p->buf_to_file && b->start == p->buf_to_file->start) {
  1071.         b->pos = p->buf_to_file->last;
  1072.         b->last = p->buf_to_file->last;

  1074.     } else {
  1075.         b->pos = b->start;
  1076.         b->last = b->start;
  1077.     }

  1079.     b->shadow = NULL;

  1081.     cl->buf = b;

  1083.     if (p->free_raw_bufs == NULL) {
  1084.         p->free_raw_bufs = cl;
  1085.         cl->next = NULL;

  1087.         return NGX_OK;
  1088.     }

  1090.     if (p->free_raw_bufs->buf->pos == p->free_raw_bufs->buf->last) {

  1092.         /* add the free buf to the list start */

  1094.         cl->next = p->free_raw_bufs;
  1095.         p->free_raw_bufs = cl;

  1097.         return NGX_OK;
  1098.     }

  1100.     /* the first free buf is partially filled, thus add the free buf after it */

  1102.     cl->next = p->free_raw_bufs->next;
  1103.     p->free_raw_bufs->next = cl;

  1105.     return NGX_OK;
  1106. }


  1109. static ngx_int_t
  1110. ‌ngx_event_pipe_drain_chains(ngx_event_pipe_t *p)
  1111. {
  1112.     ngx_chain_t  *cl, *tl;

  1114.     for ( ;; ) {
  1115.         if (p->out) {
  1116.             cl = p->out;
  1117.             p->out = NULL;

  1119.         } else if (p->in) {
  1120.             cl = p->in;
  1121.             p->in = NULL;

  1123.         } else {
  1124.             return NGX_OK;
  1125.         }

  1127.         while (cl) {
  1128.             if (cl->buf->last_shadow) {
  1129.                 if (ngx_event_pipe_add_free_buf(p, cl->buf->shadow) != NGX_OK) {
  1130.                     return NGX_ABORT;
  1131.                 }

  1133.                 cl->buf->last_shadow = 0;
  1134.             }

  1136.             cl->buf->shadow = NULL;
  1137.             tl = cl->next;
  1138.             cl->next = p->free;
  1139.             p->free = cl;
  1140.             cl = tl;
  1141.         }
  1142.     }
  1143. }

One Level Up Top Level