One Level Up Top Level

src/event/ngx_event_pipe.c - nginx source code

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->data == p->output_ctx
  257.                        && p->downstream->write->ready
  258.                        && !p->downstream->write->delayed)
  259.             {
  260.                 /*
  261.                  * if the bufs are not needed to be saved in a cache and
  262.                  * a downstream is ready then write the bufs to a downstream
  263.                  */

  265.                 p->upstream_blocked = 1;

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

  270.                 break;

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

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

  281.                 rc = ngx_event_pipe_write_chain_to_temp_file(p);

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

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

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

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

  302.             } else {

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

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

  309.                 break;
  310.             }

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

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

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

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

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

  332.                 break;
  333.             }

  335.             p->read = 1;

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

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

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

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

  351.             ngx_event_pipe_remove_shadow_links(cl->buf);

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

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

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

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

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

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

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

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

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

  389. #if (NGX_DEBUG)

  391.     for (cl = p->busy; cl; cl = cl->next) {
  392.         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
  393.                        "pipe buf busy 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->out; cl; cl = cl->next) {
  405.         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
  406.                        "pipe buf out  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->in; cl; cl = cl->next) {
  418.         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
  419.                        "pipe buf in   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.     for (cl = p->free_raw_bufs; cl; cl = cl->next) {
  431.         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
  432.                        "pipe buf free s:%d t:%d f:%d "
  433.                        "%p, pos %p, size: %z "
  434.                        "file: %O, size: %O",
  435.                        (cl->buf->shadow ? 1 : 0),
  436.                        cl->buf->temporary, cl->buf->in_file,
  437.                        cl->buf->start, cl->buf->pos,
  438.                        cl->buf->last - cl->buf->pos,
  439.                        cl->buf->file_pos,
  440.                        cl->buf->file_last - cl->buf->file_pos);
  441.     }

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

  446. #endif

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

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

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

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

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

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

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

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

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

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

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

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

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

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

  494.         rc = ngx_event_pipe_write_chain_to_temp_file(p);

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

  501.     return NGX_OK;
  502. }


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

  515.     downstream = p->downstream;

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

  520. #if (NGX_THREADS)

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

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

  530. #endif

  532.     flushed = 0;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

  590.             /* TODO: free unused bufs */

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

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

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

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

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

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

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

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

  623.         out = NULL;

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

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

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

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

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

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

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

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

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

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

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

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

  671.             cl->next = NULL;

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

  681.     flush:

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

  686.         if (out == NULL) {

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

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

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

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

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

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

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

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

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

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

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

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

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

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

  737.     return NGX_OK;
  738. }


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

  749. #if (NGX_THREADS)

  751.     if (p->writing) {

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

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

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

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

  766.         goto done;
  767.     }

  769. #endif

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

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

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

  784.     if (!p->cacheable) {

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

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

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

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

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

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

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

  816.         } while (cl);

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

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

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

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

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

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

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

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

  853. #if (NGX_THREADS)

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

  861. done:

  863. #endif

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

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

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

  878.             b = cl->buf;

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

  886.             last_out = &cl->next;

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

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

  897.         b = cl->buf;

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

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

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

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

  911.         *last_out = cl;
  912.     }

  914. free:

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

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

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

  929.         b = cl->buf;

  931.         if (b->last_shadow) {

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

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

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

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

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

  951.     return NGX_OK;
  952. }


  955. /* the copy input filter */

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

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

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

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

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

  980.         return NGX_OK;
  981.     }

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

  988.     b = cl->buf;

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

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

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

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

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

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

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

  1019.         return NGX_OK;
  1020.     }

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

  1024.     return NGX_OK;
  1025. }


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

  1033.     b = buf->shadow;

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

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

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

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

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

  1053.     b->shadow = NULL;

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


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

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

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

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

  1078.     b->shadow = NULL;

  1080.     cl->buf = b;

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

  1086.         return NGX_OK;
  1087.     }

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

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

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

  1096.         return NGX_OK;
  1097.     }

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

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

  1104.     return NGX_OK;
  1105. }


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

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

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

  1122.         } else if (p->in) {
  1123.             cl = p->in;
  1124.             p->in = NULL;

  1126.         } else {
  1127.             return NGX_OK;
  1128.         }

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

  1136.                 cl->buf->last_shadow = 0;
  1137.             }

  1139.             cl->buf->shadow = NULL;
  1140.             tl = cl->next;
  1141.             cl->next = p->free;
  1142.             p->free = cl;
  1143.             cl = tl;
  1144.         }
  1145.     }
  1146. }

One Level Up Top Level