One Level Up Top Level

src/event/ngx_event_pipe.c - nginx-1.7.10

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.     u_int         flags;
  26.     ngx_int_t     rc;
  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->fd != (ngx_socket_t) -1) {
  61.         rev = p->upstream->read;

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

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

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

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

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

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

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

  97.     return NGX_OK;
  98. }


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

  111.     if (p->upstream_eof || p->upstream_error || p->upstream_done) {
  112.         return NGX_OK;
  113.     }

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

  118.     for ( ;; ) {

  120.         if (p->upstream_eof || p->upstream_error || p->upstream_done) {
  121.             break;
  122.         }

  124.         if (p->preread_bufs == NULL && !p->upstream->read->ready) {
  125.             break;
  126.         }

  128.         if (p->preread_bufs) {

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

  132.             chain = p->preread_bufs;
  133.             p->preread_bufs = NULL;
  134.             n = p->preread_size;

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

  139.             if (n) {
  140.                 p->read = 1;
  141.             }

  143.         } else {

  145. #if (NGX_HAVE_KQUEUE)

  147.             /*
  148.              * kqueue notifies about the end of file or a pending error.
  149.              * This test allows not to allocate a buf on these conditions
  150.              * and not to call c->recv_chain().
  151.              */

  153.             if (p->upstream->read->available == 0
  154.                 && p->upstream->read->pending_eof)
  155.             {
  156.                 p->upstream->read->ready = 0;
  157.                 p->upstream->read->eof = 1;
  158.                 p->upstream_eof = 1;
  159.                 p->read = 1;

  161.                 if (p->upstream->read->kq_errno) {
  162.                     p->upstream->read->error = 1;
  163.                     p->upstream_error = 1;
  164.                     p->upstream_eof = 0;

  166.                     ngx_log_error(NGX_LOG_ERR, p->log,
  167.                                   p->upstream->read->kq_errno,
  168.                                   "kevent() reported that upstream "
  169.                                   "closed connection");
  170.                 }

  172.                 break;
  173.             }
  174. #endif

  176.             if (p->limit_rate) {
  177.                 if (p->upstream->read->delayed) {
  178.                     break;
  179.                 }

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

  184.                 if (limit <= 0) {
  185.                     p->upstream->read->delayed = 1;
  186.                     delay = (ngx_msec_t) (- limit * 1000 / p->limit_rate + 1);
  187.                     ngx_add_timer(p->upstream->read, delay);
  188.                     break;
  189.                 }

  191.             } else {
  192.                 limit = 0;
  193.             }

  195.             if (p->free_raw_bufs) {

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

  199.                 chain = p->free_raw_bufs;
  200.                 if (p->single_buf) {
  201.                     p->free_raw_bufs = p->free_raw_bufs->next;
  202.                     chain->next = NULL;
  203.                 } else {
  204.                     p->free_raw_bufs = NULL;
  205.                 }

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

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

  211.                 b = ngx_create_temp_buf(p->pool, p->bufs.size);
  212.                 if (b == NULL) {
  213.                     return NGX_ABORT;
  214.                 }

  216.                 p->allocated++;

  218.                 chain = ngx_alloc_chain_link(p->pool);
  219.                 if (chain == NULL) {
  220.                     return NGX_ABORT;
  221.                 }

  223.                 chain->buf = b;
  224.                 chain->next = NULL;

  226.             } else if (!p->cacheable
  227.                        && p->downstream->data == p->output_ctx
  228.                        && p->downstream->write->ready
  229.                        && !p->downstream->write->delayed)
  230.             {
  231.                 /*
  232.                  * if the bufs are not needed to be saved in a cache and
  233.                  * a downstream is ready then write the bufs to a downstream
  234.                  */

  236.                 p->upstream_blocked = 1;

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

  241.                 break;

  243.             } else if (p->cacheable
  244.                        || p->temp_file->offset < p->max_temp_file_size)
  245.             {

  247.                 /*
  248.                  * if it is allowed, then save some bufs from p->in
  249.                  * to a temporary file, and add them to a p->out chain
  250.                  */

  252.                 rc = ngx_event_pipe_write_chain_to_temp_file(p);

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

  257.                 if (rc == NGX_BUSY) {
  258.                     break;
  259.                 }

  261.                 if (rc == NGX_AGAIN) {
  262.                     if (ngx_event_flags & NGX_USE_LEVEL_EVENT
  263.                         && p->upstream->read->active
  264.                         && p->upstream->read->ready)
  265.                     {
  266.                         if (ngx_del_event(p->upstream->read, NGX_READ_EVENT, 0)
  267.                             == NGX_ERROR)
  268.                         {
  269.                             return NGX_ABORT;
  270.                         }
  271.                     }
  272.                 }

  274.                 if (rc != NGX_OK) {
  275.                     return rc;
  276.                 }

  278.                 chain = p->free_raw_bufs;
  279.                 if (p->single_buf) {
  280.                     p->free_raw_bufs = p->free_raw_bufs->next;
  281.                     chain->next = NULL;
  282.                 } else {
  283.                     p->free_raw_bufs = NULL;
  284.                 }

  286.             } else {

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

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

  293.                 break;
  294.             }

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

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

  301.             if (p->free_raw_bufs) {
  302.                 chain->next = p->free_raw_bufs;
  303.             }
  304.             p->free_raw_bufs = chain;

  306.             if (n == NGX_ERROR) {
  307.                 p->upstream_error = 1;
  308.                 return NGX_ERROR;
  309.             }

  311.             if (n == NGX_AGAIN) {
  312.                 if (p->single_buf) {
  313.                     ngx_event_pipe_remove_shadow_links(chain->buf);
  314.                 }

  316.                 break;
  317.             }

  319.             p->read = 1;

  321.             if (n == 0) {
  322.                 p->upstream_eof = 1;
  323.                 break;
  324.             }
  325.         }

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

  329.         p->read_length += n;
  330.         cl = chain;
  331.         p->free_raw_bufs = NULL;

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

  335.             ngx_event_pipe_remove_shadow_links(cl->buf);

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

  339.             if (n >= size) {
  340.                 cl->buf->last = cl->buf->end;

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

  344.                 if (p->input_filter(p, cl->buf) == NGX_ERROR) {
  345.                     return NGX_ABORT;
  346.                 }

  348.                 n -= size;
  349.                 ln = cl;
  350.                 cl = cl->next;
  351.                 ngx_free_chain(p->pool, ln);

  353.             } else {
  354.                 cl->buf->last += n;
  355.                 n = 0;
  356.             }
  357.         }

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

  362.             ln->next = p->free_raw_bufs;
  363.             p->free_raw_bufs = cl;
  364.         }

  366.         if (delay > 0) {
  367.             p->upstream->read->delayed = 1;
  368.             ngx_add_timer(p->upstream->read, delay);
  369.             break;
  370.         }
  371.     }

  373. #if (NGX_DEBUG)

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

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

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

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

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

  430. #endif

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

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

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

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

  441.             if (p->input_filter(p, cl->buf) == NGX_ERROR) {
  442.                  return NGX_ABORT;
  443.             }

  445.             ngx_free_chain(p->pool, cl);
  446.         }
  447.     }

  449.     if (p->length == 0) {
  450.         p->upstream_done = 1;
  451.         p->read = 1;
  452.     }

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

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

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

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

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

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

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

  478.         if (ngx_event_pipe_write_chain_to_temp_file(p) == NGX_ABORT) {
  479.             return NGX_ABORT;
  480.         }
  481.     }

  483.     return NGX_OK;
  484. }


  487. static ngx_int_t
  488. ‌ngx_event_pipe_write_to_downstream(ngx_event_pipe_t *p)
  489. {
  490.     u_char            *prev;
  491.     size_t             bsize;
  492.     ngx_int_t          rc;
  493.     ngx_uint_t         flush, flushed, prev_last_shadow;
  494.     ngx_chain_t       *out, **ll, *cl;
  495.     ngx_connection_t  *downstream;

  497.     downstream = p->downstream;

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

  502.     flushed = 0;

  504.     for ( ;; ) {
  505.         if (p->downstream_error) {
  506.             return ngx_event_pipe_drain_chains(p);
  507.         }

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

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

  513.             for (cl = p->busy; cl; cl = cl->next) {
  514.                 cl->buf->recycled = 0;
  515.             }

  517.             if (p->out) {
  518.                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
  519.                                "pipe write downstream flush out");

  521.                 for (cl = p->out; cl; cl = cl->next) {
  522.                     cl->buf->recycled = 0;
  523.                 }

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

  527.                 if (rc == NGX_ERROR) {
  528.                     p->downstream_error = 1;
  529.                     return ngx_event_pipe_drain_chains(p);
  530.                 }

  532.                 p->out = NULL;
  533.             }

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

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

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

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

  550.                 p->in = NULL;
  551.             }

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

  556.             /* TODO: free unused bufs */

  558.             p->downstream_done = 1;
  559.             break;
  560.         }

  562.         if (downstream->data != p->output_ctx
  563.             || !downstream->write->ready
  564.             || downstream->write->delayed)
  565.         {
  566.             break;
  567.         }

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

  571.         prev = NULL;
  572.         bsize = 0;

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

  576.             if (cl->buf->recycled) {
  577.                 if (prev == cl->buf->start) {
  578.                     continue;
  579.                 }

  581.                 bsize += cl->buf->end - cl->buf->start;
  582.                 prev = cl->buf->start;
  583.             }
  584.         }

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

  589.         out = NULL;

  591.         if (bsize >= (size_t) p->busy_size) {
  592.             flush = 1;
  593.             goto flush;
  594.         }

  596.         flush = 0;
  597.         ll = NULL;
  598.         prev_last_shadow = 1;

  600.         for ( ;; ) {
  601.             if (p->out) {
  602.                 cl = p->out;

  604.                 if (cl->buf->recycled) {
  605.                     ngx_log_error(NGX_LOG_ALERT, p->log, 0,
  606.                                   "recycled buffer in pipe out chain");
  607.                 }

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

  611.             } else if (!p->cacheable && p->in) {
  612.                 cl = p->in;

  614.                 ngx_log_debug3(NGX_LOG_DEBUG_EVENT, p->log, 0,
  615.                                "pipe write buf ls:%d %p %z",
  616.                                cl->buf->last_shadow,
  617.                                cl->buf->pos,
  618.                                cl->buf->last - cl->buf->pos);

  620.                 if (cl->buf->recycled && prev_last_shadow) {
  621.                     if (bsize + cl->buf->end - cl->buf->start > p->busy_size) {
  622.                         flush = 1;
  623.                         break;
  624.                     }

  626.                     bsize += cl->buf->end - cl->buf->start;
  627.                 }

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

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

  633.             } else {
  634.                 break;
  635.             }

  637.             cl->next = NULL;

  639.             if (out) {
  640.                 *ll = cl;
  641.             } else {
  642.                 out = cl;
  643.             }
  644.             ll = &cl->next;
  645.         }

  647.     flush:

  649.         ngx_log_debug2(NGX_LOG_DEBUG_EVENT, p->log, 0,
  650.                        "pipe write: out:%p, f:%d", out, flush);

  652.         if (out == NULL) {

  654.             if (!flush) {
  655.                 break;
  656.             }

  658.             /* a workaround for AIO */
  659.             if (flushed++ > 10) {
  660.                 return NGX_BUSY;
  661.             }
  662.         }

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

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

  668.         if (rc == NGX_ERROR) {
  669.             p->downstream_error = 1;
  670.             return ngx_event_pipe_drain_chains(p);
  671.         }

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

  675.             if (cl->buf->temp_file) {
  676.                 if (p->cacheable || !p->cyclic_temp_file) {
  677.                     continue;
  678.                 }

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

  682.                 if (cl->buf->file_last == p->temp_file->offset) {
  683.                     p->temp_file->offset = 0;
  684.                 }
  685.             }

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

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

  691.             if (cl->buf->last_shadow) {
  692.                 if (ngx_event_pipe_add_free_buf(p, cl->buf->shadow) != NGX_OK) {
  693.                     return NGX_ABORT;
  694.                 }

  696.                 cl->buf->last_shadow = 0;
  697.             }

  699.             cl->buf->shadow = NULL;
  700.         }
  701.     }

  703.     return NGX_OK;
  704. }


  707. static ngx_int_t
  708. ‌ngx_event_pipe_write_chain_to_temp_file(ngx_event_pipe_t *p)
  709. {
  710.     ssize_t       size, bsize, n;
  711.     ngx_buf_t    *b;
  712.     ngx_uint_t    prev_last_shadow;
  713.     ngx_chain_t  *cl, *tl, *next, *out, **ll, **last_out, **last_free, fl;

  715.     if (p->buf_to_file) {
  716.         fl.buf = p->buf_to_file;
  717.         fl.next = p->in;
  718.         out = &fl;

  720.     } else {
  721.         out = p->in;
  722.     }

  724.     if (!p->cacheable) {

  726.         size = 0;
  727.         cl = out;
  728.         ll = NULL;
  729.         prev_last_shadow = 1;

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

  734.         do {
  735.             bsize = cl->buf->last - cl->buf->pos;

  737.             ngx_log_debug4(NGX_LOG_DEBUG_EVENT, p->log, 0,
  738.                            "pipe buf ls:%d %p, pos %p, size: %z",
  739.                            cl->buf->last_shadow, cl->buf->start,
  740.                            cl->buf->pos, bsize);

  742.             if (prev_last_shadow
  743.                 && ((size + bsize > p->temp_file_write_size)
  744.                     || (p->temp_file->offset + size + bsize
  745.                         > p->max_temp_file_size)))
  746.             {
  747.                 break;
  748.             }

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

  752.             size += bsize;
  753.             ll = &cl->next;
  754.             cl = cl->next;

  756.         } while (cl);

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

  760.         if (ll == NULL) {
  761.             return NGX_BUSY;
  762.         }

  764.         if (cl) {
  765.            p->in = cl;
  766.            *ll = NULL;

  768.         } else {
  769.            p->in = NULL;
  770.            p->last_in = &p->in;
  771.         }

  773.     } else {
  774.         p->in = NULL;
  775.         p->last_in = &p->in;
  776.     }

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

  780.     if (n == NGX_ERROR) {
  781.         return NGX_ABORT;
  782.     }

  784.     if (p->buf_to_file) {
  785.         p->temp_file->offset = p->buf_to_file->last - p->buf_to_file->pos;
  786.         n -= p->buf_to_file->last - p->buf_to_file->pos;
  787.         p->buf_to_file = NULL;
  788.         out = out->next;
  789.     }

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

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

  797.             b = cl->buf;

  799.             if (b->file_last == p->temp_file->offset) {
  800.                 p->temp_file->offset += n;
  801.                 b->file_last = p->temp_file->offset;
  802.                 goto free;
  803.             }

  805.             last_out = &cl->next;

  807.         } else {
  808.             last_out = &p->out;
  809.         }

  811.         cl = ngx_chain_get_free_buf(p->pool, &p->free);
  812.         if (cl == NULL) {
  813.             return NGX_ABORT;
  814.         }

  816.         b = cl->buf;

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

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

  822.         b->file = &p->temp_file->file;
  823.         b->file_pos = p->temp_file->offset;
  824.         p->temp_file->offset += n;
  825.         b->file_last = p->temp_file->offset;

  827.         b->in_file = 1;
  828.         b->temp_file = 1;

  830.         *last_out = cl;
  831.     }

  833. free:

  835.     for (last_free = &p->free_raw_bufs;
  836.          *last_free != NULL;
  837.          last_free = &(*last_free)->next)
  838.     {
  839.         /* void */
  840.     }

  842.     for (cl = out; cl; cl = next) {
  843.         next = cl->next;

  845.         cl->next = p->free;
  846.         p->free = cl;

  848.         b = cl->buf;

  850.         if (b->last_shadow) {

  852.             tl = ngx_alloc_chain_link(p->pool);
  853.             if (tl == NULL) {
  854.                 return NGX_ABORT;
  855.             }

  857.             tl->buf = b->shadow;
  858.             tl->next = NULL;

  860.             *last_free = tl;
  861.             last_free = &tl->next;

  863.             b->shadow->pos = b->shadow->start;
  864.             b->shadow->last = b->shadow->start;

  866.             ngx_event_pipe_remove_shadow_links(b->shadow);
  867.         }
  868.     }

  870.     return NGX_OK;
  871. }


  874. /* the copy input filter */

  876. ngx_int_t
  877. ‌ngx_event_pipe_copy_input_filter(ngx_event_pipe_t *p, ngx_buf_t *buf)
  878. {
  879.     ngx_buf_t    *b;
  880.     ngx_chain_t  *cl;

  882.     if (buf->pos == buf->last) {
  883.         return NGX_OK;
  884.     }

  886.     cl = ngx_chain_get_free_buf(p->pool, &p->free);
  887.     if (cl == NULL) {
  888.         return NGX_ERROR;
  889.     }

  891.     b = cl->buf;

  893.     ngx_memcpy(b, buf, sizeof(ngx_buf_t));
  894.     b->shadow = buf;
  895.     b->tag = p->tag;
  896.     b->last_shadow = 1;
  897.     b->recycled = 1;
  898.     buf->shadow = b;

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

  902.     if (p->in) {
  903.         *p->last_in = cl;
  904.     } else {
  905.         p->in = cl;
  906.     }
  907.     p->last_in = &cl->next;

  909.     if (p->length == -1) {
  910.         return NGX_OK;
  911.     }

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

  915.     return NGX_OK;
  916. }


  919. static ngx_inline void
  920. ‌ngx_event_pipe_remove_shadow_links(ngx_buf_t *buf)
  921. {
  922.     ngx_buf_t  *b, *next;

  924.     b = buf->shadow;

  926.     if (b == NULL) {
  927.         return;
  928.     }

  930.     while (!b->last_shadow) {
  931.         next = b->shadow;

  933.         b->temporary = 0;
  934.         b->recycled = 0;

  936.         b->shadow = NULL;
  937.         b = next;
  938.     }

  940.     b->temporary = 0;
  941.     b->recycled = 0;
  942.     b->last_shadow = 0;

  944.     b->shadow = NULL;

  946.     buf->shadow = NULL;
  947. }


  950. ngx_int_t
  951. ‌ngx_event_pipe_add_free_buf(ngx_event_pipe_t *p, ngx_buf_t *b)
  952. {
  953.     ngx_chain_t  *cl;

  955.     cl = ngx_alloc_chain_link(p->pool);
  956.     if (cl == NULL) {
  957.         return NGX_ERROR;
  958.     }

  960.     if (p->buf_to_file && b->start == p->buf_to_file->start) {
  961.         b->pos = p->buf_to_file->last;
  962.         b->last = p->buf_to_file->last;

  964.     } else {
  965.         b->pos = b->start;
  966.         b->last = b->start;
  967.     }

  969.     b->shadow = NULL;

  971.     cl->buf = b;

  973.     if (p->free_raw_bufs == NULL) {
  974.         p->free_raw_bufs = cl;
  975.         cl->next = NULL;

  977.         return NGX_OK;
  978.     }

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

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

  984.         cl->next = p->free_raw_bufs;
  985.         p->free_raw_bufs = cl;

  987.         return NGX_OK;
  988.     }

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

  992.     cl->next = p->free_raw_bufs->next;
  993.     p->free_raw_bufs->next = cl;

  995.     return NGX_OK;
  996. }


  999. static ngx_int_t
  1000. ‌ngx_event_pipe_drain_chains(ngx_event_pipe_t *p)
  1001. {
  1002.     ngx_chain_t  *cl, *tl;

  1004.     for ( ;; ) {
  1005.         if (p->busy) {
  1006.             cl = p->busy;
  1007.             p->busy = NULL;

  1009.         } else if (p->out) {
  1010.             cl = p->out;
  1011.             p->out = NULL;

  1013.         } else if (p->in) {
  1014.             cl = p->in;
  1015.             p->in = NULL;

  1017.         } else {
  1018.             return NGX_OK;
  1019.         }

  1021.         while (cl) {
  1022.             if (cl->buf->last_shadow) {
  1023.                 if (ngx_event_pipe_add_free_buf(p, cl->buf->shadow) != NGX_OK) {
  1024.                     return NGX_ABORT;
  1025.                 }

  1027.                 cl->buf->last_shadow = 0;
  1028.             }

  1030.             cl->buf->shadow = NULL;
  1031.             tl = cl->next;
  1032.             cl->next = p->free;
  1033.             p->free = cl;
  1034.             cl = tl;
  1035.         }
  1036.     }
  1037. }

One Level Up Top Level