One Level Up Top Level

src/event/ngx_event_pipe.c - nginx-1.31.9 nginx/ @ 2a3a19c6f

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.                 p->free_raw_bufs = NULL;

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

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

  235.                 b = ngx_create_temp_buf(p->pool, p->bufs.size);
  236.                 if (b == NULL) {
  237.                     return NGX_ABORT;
  238.                 }

  240.                 p->allocated++;

  242.                 chain = ngx_alloc_chain_link(p->pool);
  243.                 if (chain == NULL) {
  244.                     return NGX_ABORT;
  245.                 }

  247.                 chain->buf = b;
  248.                 chain->next = NULL;

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

  261.                 p->upstream_blocked = 1;

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

  266.                 break;

  268.             } else if (p->cacheable
  269.                        || p->temp_file->offset < p->max_temp_file_size)
  270.             {

  272.                 /*
  273.                  * if it is allowed, then save some bufs from p->in
  274.                  * to a temporary file, and add them to a p->out chain
  275.                  */

  277.                 rc = ngx_event_pipe_write_chain_to_temp_file(p);

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

  282.                 if (rc == NGX_BUSY) {
  283.                     break;
  284.                 }

  286.                 if (rc != NGX_OK) {
  287.                     return rc;
  288.                 }

  290.                 chain = p->free_raw_bufs;
  291.                 p->free_raw_bufs = NULL;

  293.             } else {

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

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

  300.                 break;
  301.             }

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

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

  308.             if (p->free_raw_bufs) {
  309.                 chain->next = p->free_raw_bufs;
  310.             }
  311.             p->free_raw_bufs = chain;

  313.             if (n == NGX_ERROR) {
  314.                 p->upstream_error = 1;
  315.                 break;
  316.             }

  318.             if (n == NGX_AGAIN) {
  319.                 break;
  320.             }

  322.             p->read = 1;

  324.             if (n == 0) {
  325.                 p->upstream_eof = 1;
  326.                 break;
  327.             }
  328.         }

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

  332.         p->read_length += n;
  333.         cl = chain;
  334.         p->free_raw_bufs = NULL;

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

  338.             ngx_event_pipe_remove_shadow_links(cl->buf);

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

  342.             if (n >= size) {
  343.                 cl->buf->last = cl->buf->end;

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

  347.                 if (p->input_filter(p, cl->buf) == NGX_ERROR) {
  348.                     return NGX_ABORT;
  349.                 }

  351.                 n -= size;
  352.                 ln = cl;
  353.                 cl = cl->next;
  354.                 ngx_free_chain(p->pool, ln);

  356.             } else {
  357.                 cl->buf->last += n;
  358.                 n = 0;
  359.             }
  360.         }

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

  365.             ln->next = p->free_raw_bufs;
  366.             p->free_raw_bufs = cl;
  367.         }

  369.         if (delay > 0) {
  370.             p->upstream->read->delayed = 1;
  371.             ngx_add_timer(p->upstream->read, delay);
  372.             break;
  373.         }
  374.     }

  376. #if (NGX_DEBUG)

  378.     for (cl = p->busy; cl; cl = cl->next) {
  379.         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
  380.                        "pipe buf busy s:%d t:%d f:%d "
  381.                        "%p, pos %p, size: %z "
  382.                        "file: %O, size: %O",
  383.                        (cl->buf->shadow ? 1 : 0),
  384.                        cl->buf->temporary, cl->buf->in_file,
  385.                        cl->buf->start, cl->buf->pos,
  386.                        cl->buf->last - cl->buf->pos,
  387.                        cl->buf->file_pos,
  388.                        cl->buf->file_last - cl->buf->file_pos);
  389.     }

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

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

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

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

  433. #endif

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

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

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

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

  444.             if (p->input_filter(p, cl->buf) == NGX_ERROR) {
  445.                 return NGX_ABORT;
  446.             }

  448.             ngx_free_chain(p->pool, cl);
  449.         }
  450.     }

  452.     if (p->length == 0) {
  453.         p->upstream_done = 1;
  454.         p->read = 1;
  455.     }

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

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

  461.         if (p->input_filter(p, p->free_raw_bufs->buf) == NGX_ERROR) {
  462.             return NGX_ABORT;
  463.         }

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

  467.         if (p->free_bufs && p->buf_to_file == NULL) {
  468.             for (cl = p->free_raw_bufs; cl; cl = cl->next) {
  469.                 if (cl->buf->shadow == NULL) {
  470.                     ngx_pfree(p->pool, cl->buf->start);
  471.                 }
  472.             }
  473.         }
  474.     }

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

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

  481.         rc = ngx_event_pipe_write_chain_to_temp_file(p);

  483.         if (rc != NGX_OK) {
  484.             return rc;
  485.         }
  486.     }

  488.     return NGX_OK;
  489. }


  492. static ngx_int_t
  493. ‌ngx_event_pipe_write_to_downstream(ngx_event_pipe_t *p)
  494. {
  495.     u_char            *prev;
  496.     size_t             bsize;
  497.     ngx_int_t          rc;
  498.     ngx_uint_t         flush, flushed, prev_last_shadow;
  499.     ngx_chain_t       *out, **ll, *cl;
  500.     ngx_connection_t  *downstream;

  502.     downstream = p->downstream;

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

  507. #if (NGX_THREADS)

  509.     if (p->writing) {
  510.         rc = ngx_event_pipe_write_chain_to_temp_file(p);

  512.         if (rc == NGX_ABORT) {
  513.             return NGX_ABORT;
  514.         }
  515.     }

  517. #endif

  519.     flushed = 0;

  521.     for ( ;; ) {
  522.         if (p->downstream_error) {
  523.             return ngx_event_pipe_drain_chains(p);
  524.         }

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

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

  530.             for (cl = p->busy; cl; cl = cl->next) {
  531.                 cl->buf->recycled = 0;
  532.             }

  534.             if (p->out) {
  535.                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  536.                                "pipe write downstream flush out");

  538.                 for (cl = p->out; cl; cl = cl->next) {
  539.                     cl->buf->recycled = 0;
  540.                 }

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

  544.                 if (rc == NGX_ERROR) {
  545.                     p->downstream_error = 1;
  546.                     return ngx_event_pipe_drain_chains(p);
  547.                 }

  549.                 p->out = NULL;
  550.             }

  552.             if (p->writing) {
  553.                 break;
  554.             }

  556.             if (p->in) {
  557.                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  558.                                "pipe write downstream flush in");

  560.                 for (cl = p->in; cl; cl = cl->next) {
  561.                     cl->buf->recycled = 0;
  562.                 }

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

  566.                 if (rc == NGX_ERROR) {
  567.                     p->downstream_error = 1;
  568.                     return ngx_event_pipe_drain_chains(p);
  569.                 }

  571.                 p->in = NULL;
  572.             }

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

  577.             /* TODO: free unused bufs */

  579.             p->downstream_done = 1;
  580.             break;
  581.         }

  583.         if (downstream->data != p->output_ctx
  584.             || !downstream->write->ready
  585.             || downstream->write->delayed)
  586.         {
  587.             break;
  588.         }

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

  592.         prev = NULL;
  593.         bsize = 0;

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

  597.             if (cl->buf->recycled) {
  598.                 if (prev == cl->buf->start) {
  599.                     continue;
  600.                 }

  602.                 bsize += cl->buf->end - cl->buf->start;
  603.                 prev = cl->buf->start;
  604.             }
  605.         }

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

  610.         out = NULL;

  612.         if (bsize >= (size_t) p->busy_size) {
  613.             flush = 1;
  614.             goto flush;
  615.         }

  617.         flush = 0;
  618.         ll = NULL;
  619.         prev_last_shadow = 1;

  621.         for ( ;; ) {
  622.             if (p->out) {
  623.                 cl = p->out;

  625.                 if (cl->buf->recycled) {
  626.                     ngx_log_error(NGX_LOG_ALERT, p->log, 0,
  627.                                   "recycled buffer in pipe out chain");
  628.                 }

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

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

  635.                 ngx_log_debug3(NGX_LOG_DEBUG_EVENT, p->log, 0,
  636.                                "pipe write buf ls:%d %p %z",
  637.                                cl->buf->last_shadow,
  638.                                cl->buf->pos,
  639.                                cl->buf->last - cl->buf->pos);

  641.                 if (cl->buf->recycled && prev_last_shadow) {
  642.                     if (bsize + cl->buf->end - cl->buf->start > p->busy_size) {
  643.                         flush = 1;
  644.                         break;
  645.                     }

  647.                     bsize += cl->buf->end - cl->buf->start;
  648.                 }

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

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

  654.             } else {
  655.                 break;
  656.             }

  658.             cl->next = NULL;

  660.             if (out) {
  661.                 *ll = cl;
  662.             } else {
  663.                 out = cl;
  664.             }
  665.             ll = &cl->next;
  666.         }

  668.     flush:

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

  673.         if (out == NULL) {

  675.             if (!flush) {
  676.                 break;
  677.             }

  679.             /* a workaround for AIO */
  680.             if (flushed++ > 10) {
  681.                 return NGX_BUSY;
  682.             }
  683.         }

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

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

  689.         if (rc == NGX_ERROR) {
  690.             p->downstream_error = 1;
  691.             return ngx_event_pipe_drain_chains(p);
  692.         }

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

  696.             if (cl->buf->temp_file) {
  697.                 if (p->cacheable || !p->cyclic_temp_file) {
  698.                     continue;
  699.                 }

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

  703.                 if (cl->buf->file_last == p->temp_file->offset) {
  704.                     p->temp_file->offset = 0;
  705.                 }
  706.             }

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

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

  712.             if (cl->buf->last_shadow) {
  713.                 if (ngx_event_pipe_add_free_buf(p, cl->buf->shadow) != NGX_OK) {
  714.                     return NGX_ABORT;
  715.                 }

  717.                 cl->buf->last_shadow = 0;
  718.             }

  720.             cl->buf->shadow = NULL;
  721.         }
  722.     }

  724.     return NGX_OK;
  725. }


  728. static ngx_int_t
  729. ‌ngx_event_pipe_write_chain_to_temp_file(ngx_event_pipe_t *p)
  730. {
  731.     ssize_t       size, bsize, n;
  732.     ngx_buf_t    *b;
  733.     ngx_uint_t    prev_last_shadow;
  734.     ngx_chain_t  *cl, *tl, *next, *out, **ll, **last_out, **last_free;

  736. #if (NGX_THREADS)

  738.     if (p->writing) {

  740.         if (p->aio) {
  741.             return NGX_AGAIN;
  742.         }

  744.         out = p->writing;
  745.         p->writing = NULL;

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

  749.         if (n == NGX_ERROR) {
  750.             return NGX_ABORT;
  751.         }

  753.         goto done;
  754.     }

  756. #endif

  758.     if (p->buf_to_file) {
  759.         out = ngx_alloc_chain_link(p->pool);
  760.         if (out == NULL) {
  761.             return NGX_ABORT;
  762.         }

  764.         out->buf = p->buf_to_file;
  765.         out->next = p->in;

  767.     } else {
  768.         out = p->in;
  769.     }

  771.     if (!p->cacheable) {

  773.         size = 0;
  774.         cl = out;
  775.         ll = NULL;
  776.         prev_last_shadow = 1;

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

  781.         do {
  782.             bsize = cl->buf->last - cl->buf->pos;

  784.             ngx_log_debug4(NGX_LOG_DEBUG_EVENT, p->log, 0,
  785.                            "pipe buf ls:%d %p, pos %p, size: %z",
  786.                            cl->buf->last_shadow, cl->buf->start,
  787.                            cl->buf->pos, bsize);

  789.             if (prev_last_shadow
  790.                 && ((size + bsize > p->temp_file_write_size)
  791.                     || (p->temp_file->offset + size + bsize
  792.                         > p->max_temp_file_size)))
  793.             {
  794.                 break;
  795.             }

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

  799.             size += bsize;
  800.             ll = &cl->next;
  801.             cl = cl->next;

  803.         } while (cl);

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

  807.         if (ll == NULL) {
  808.             return NGX_BUSY;
  809.         }

  811.         if (cl) {
  812.             p->in = cl;
  813.             *ll = NULL;

  815.         } else {
  816.             p->in = NULL;
  817.             p->last_in = &p->in;
  818.         }

  820.     } else {
  821.         p->in = NULL;
  822.         p->last_in = &p->in;
  823.     }

  825. #if (NGX_THREADS)
  826.     if (p->thread_handler) {
  827.         p->temp_file->thread_write = 1;
  828.         p->temp_file->file.thread_task = p->thread_task;
  829.         p->temp_file->file.thread_handler = p->thread_handler;
  830.         p->temp_file->file.thread_ctx = p->thread_ctx;
  831.     }
  832. #endif

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

  836.     if (n == NGX_ERROR) {
  837.         return NGX_ABORT;
  838.     }

  840. #if (NGX_THREADS)

  842.     if (n == NGX_AGAIN) {
  843.         p->writing = out;
  844.         p->thread_task = p->temp_file->file.thread_task;
  845.         return NGX_AGAIN;
  846.     }

  848. done:

  850. #endif

  852.     if (p->buf_to_file) {
  853.         p->temp_file->offset = p->buf_to_file->last - p->buf_to_file->pos;
  854.         n -= p->buf_to_file->last - p->buf_to_file->pos;
  855.         p->buf_to_file = NULL;
  856.         out = out->next;
  857.     }

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

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

  865.             b = cl->buf;

  867.             if (b->file_last == p->temp_file->offset) {
  868.                 p->temp_file->offset += n;
  869.                 b->file_last = p->temp_file->offset;
  870.                 goto free;
  871.             }

  873.             last_out = &cl->next;

  875.         } else {
  876.             last_out = &p->out;
  877.         }

  879.         cl = ngx_chain_get_free_buf(p->pool, &p->free);
  880.         if (cl == NULL) {
  881.             return NGX_ABORT;
  882.         }

  884.         b = cl->buf;

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

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

  890.         b->file = &p->temp_file->file;
  891.         b->file_pos = p->temp_file->offset;
  892.         p->temp_file->offset += n;
  893.         b->file_last = p->temp_file->offset;

  895.         b->in_file = 1;
  896.         b->temp_file = 1;

  898.         *last_out = cl;
  899.     }

  901. free:

  903.     for (last_free = &p->free_raw_bufs;
  904.          *last_free != NULL;
  905.          last_free = &(*last_free)->next)
  906.     {
  907.         /* void */
  908.     }

  910.     for (cl = out; cl; cl = next) {
  911.         next = cl->next;

  913.         cl->next = p->free;
  914.         p->free = cl;

  916.         b = cl->buf;

  918.         if (b->last_shadow) {

  920.             tl = ngx_alloc_chain_link(p->pool);
  921.             if (tl == NULL) {
  922.                 return NGX_ABORT;
  923.             }

  925.             tl->buf = b->shadow;
  926.             tl->next = NULL;

  928.             *last_free = tl;
  929.             last_free = &tl->next;

  931.             b->shadow->pos = b->shadow->start;
  932.             b->shadow->last = b->shadow->start;

  934.             ngx_event_pipe_remove_shadow_links(b->shadow);
  935.         }
  936.     }

  938.     return NGX_OK;
  939. }


  942. /* the copy input filter */

  944. ngx_int_t
  945. ‌ngx_event_pipe_copy_input_filter(ngx_event_pipe_t *p, ngx_buf_t *buf)
  946. {
  947.     ngx_buf_t    *b;
  948.     ngx_chain_t  *cl;

  950.     if (buf->pos == buf->last) {
  951.         return NGX_OK;
  952.     }

  954.     if (p->upstream_done) {
  955.         ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  956.                        "input data after close");
  957.         return NGX_OK;
  958.     }

  960.     if (p->length == 0) {
  961.         p->upstream_done = 1;

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

  967.         return NGX_OK;
  968.     }

  970.     cl = ngx_chain_get_free_buf(p->pool, &p->free);
  971.     if (cl == NULL) {
  972.         return NGX_ERROR;
  973.     }

  975.     b = cl->buf;

  977.     ngx_memcpy(b, buf, sizeof(ngx_buf_t));
  978.     b->shadow = buf;
  979.     b->tag = p->tag;
  980.     b->last_shadow = 1;
  981.     b->recycled = 1;
  982.     buf->shadow = b;

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

  986.     if (p->in) {
  987.         *p->last_in = cl;
  988.     } else {
  989.         p->in = cl;
  990.     }
  991.     p->last_in = &cl->next;

  993.     if (p->length == -1) {
  994.         return NGX_OK;
  995.     }

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

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

  1003.         b->last = b->pos + p->length;
  1004.         p->upstream_done = 1;

  1006.         return NGX_OK;
  1007.     }

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

  1011.     return NGX_OK;
  1012. }


  1015. static ngx_inline void
  1016. ‌ngx_event_pipe_remove_shadow_links(ngx_buf_t *buf)
  1017. {
  1018.     ngx_buf_t  *b, *next;

  1020.     b = buf->shadow;

  1022.     if (b == NULL) {
  1023.         return;
  1024.     }

  1026.     while (!b->last_shadow) {
  1027.         next = b->shadow;

  1029.         b->temporary = 0;
  1030.         b->recycled = 0;

  1032.         b->shadow = NULL;
  1033.         b = next;
  1034.     }

  1036.     b->temporary = 0;
  1037.     b->recycled = 0;
  1038.     b->last_shadow = 0;

  1040.     b->shadow = NULL;

  1042.     buf->shadow = NULL;
  1043. }


  1046. ngx_int_t
  1047. ‌ngx_event_pipe_add_free_buf(ngx_event_pipe_t *p, ngx_buf_t *b)
  1048. {
  1049.     ngx_chain_t  *cl;

  1051.     cl = ngx_alloc_chain_link(p->pool);
  1052.     if (cl == NULL) {
  1053.         return NGX_ERROR;
  1054.     }

  1056.     if (p->buf_to_file && b->start == p->buf_to_file->start) {
  1057.         b->pos = p->buf_to_file->last;
  1058.         b->last = p->buf_to_file->last;

  1060.     } else {
  1061.         b->pos = b->start;
  1062.         b->last = b->start;
  1063.     }

  1065.     b->shadow = NULL;

  1067.     cl->buf = b;

  1069.     if (p->free_raw_bufs == NULL) {
  1070.         p->free_raw_bufs = cl;
  1071.         cl->next = NULL;

  1073.         return NGX_OK;
  1074.     }

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

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

  1080.         cl->next = p->free_raw_bufs;
  1081.         p->free_raw_bufs = cl;

  1083.         return NGX_OK;
  1084.     }

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

  1088.     cl->next = p->free_raw_bufs->next;
  1089.     p->free_raw_bufs->next = cl;

  1091.     return NGX_OK;
  1092. }


  1095. static ngx_int_t
  1096. ‌ngx_event_pipe_drain_chains(ngx_event_pipe_t *p)
  1097. {
  1098.     ngx_chain_t  *cl, *tl;

  1100.     for ( ;; ) {
  1101.         if (p->out) {
  1102.             cl = p->out;
  1103.             p->out = NULL;

  1105.         } else if (p->in) {
  1106.             cl = p->in;
  1107.             p->in = NULL;

  1109.         } else {
  1110.             return NGX_OK;
  1111.         }

  1113.         while (cl) {
  1114.             if (cl->buf->last_shadow) {
  1115.                 if (ngx_event_pipe_add_free_buf(p, cl->buf->shadow) != NGX_OK) {
  1116.                     return NGX_ABORT;
  1117.                 }

  1119.                 cl->buf->last_shadow = 0;
  1120.             }

  1122.             cl->buf->shadow = NULL;
  1123.             tl = cl->next;
  1124.             cl->next = p->free;
  1125.             p->free = cl;
  1126.             cl = tl;
  1127.         }
  1128.     }
  1129. }

One Level Up Top Level