nginx/src/event/ngx_event_pipe.c

   1
   2 /*
   3  * Copyright (C) Igor Sysoev
   4  */
   5
   6
   7 #include <ngx_config.h>
   8 #include <ngx_core.h>
   9 #include <ngx_event.h>
  10 #include <ngx_event_pipe.h>
  11
  12
  13 static ngx_int_t ngx_event_pipe_read_upstream(ngx_event_pipe_t *p);
  14 static ngx_int_t ngx_event_pipe_write_to_downstream(ngx_event_pipe_t *p);
  15
  16 static ngx_int_t ngx_event_pipe_write_chain_to_temp_file(ngx_event_pipe_t *p);
  17 static ngx_inline void ngx_event_pipe_remove_shadow_links(ngx_buf_t *buf);
  18 static ngx_inline void ngx_event_pipe_free_shadow_raw_buf(ngx_chain_t **free,
  19                                                           ngx_buf_t *buf);
  20 static ngx_int_t ngx_event_pipe_drain_chains(ngx_event_pipe_t *p);
  21
  22
  23 ngx_int_t
  24 ngx_event_pipe(ngx_event_pipe_t *p, ngx_int_t do_write)
  25 {
  26     u_int         flags;
  27     ngx_event_t  *rev, *wev;
  28
  29     for ( ;; ) {
  30         if (do_write) {
  31             p->log->action = "sending to client";
  32
  33             if (ngx_event_pipe_write_to_downstream(p) == NGX_ABORT) {
  34                 return NGX_ABORT;
  35             }
  36         }
  37
  38         p->read = 0;
  39         p->upstream_blocked = 0;
  40
  41         p->log->action = "reading upstream";
  42
  43         if (ngx_event_pipe_read_upstream(p) == NGX_ABORT) {
  44             return NGX_ABORT;
  45         }
  46
  47         if (!p->read && !p->upstream_blocked) {
  48             break;
  49         }
  50
  51         do_write = 1;
  52     }
  53
  54     if (p->upstream->fd != -1) {
  55         rev = p->upstream->read;
  56
  57         flags = (rev->eof || rev->error) ? NGX_CLOSE_EVENT : 0;
  58
  59         if (ngx_handle_read_event(rev, flags) != NGX_OK) {
  60             return NGX_ABORT;
  61         }
  62
  63         if (rev->active && !rev->ready) {
  64             ngx_add_timer(rev, p->read_timeout);
  65
  66         } else if (rev->timer_set) {
  67             ngx_del_timer(rev);
  68         }
  69     }
  70
  71     if (p->downstream->fd != -1 && p->downstream->data == p->output_ctx) {
  72         wev = p->downstream->write;
  73         if (ngx_handle_write_event(wev, p->send_lowat) != NGX_OK) {
  74             return NGX_ABORT;
  75         }
  76
  77         if (!wev->delayed) {
  78             if (wev->active && !wev->ready) {
  79                 ngx_add_timer(wev, p->send_timeout);
  80
  81             } else if (wev->timer_set) {
  82                 ngx_del_timer(wev);
  83             }
  84         }
  85     }
  86
  87     return NGX_OK;
  88 }
  89
  90
  91 static ngx_int_t
  92 ngx_event_pipe_read_upstream(ngx_event_pipe_t *p)
  93 {
  94     ssize_t       n, size;
  95     ngx_int_t     rc;
  96     ngx_buf_t    *b;
  97     ngx_chain_t  *chain, *cl, *ln;
  98
  99     if (p->upstream_eof || p->upstream_error || p->upstream_done) {
 100         return NGX_OK;
 101     }
 102
 103     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
 104                    "pipe read upstream: %d", p->upstream->read->ready);
 105
 106     for ( ;; ) {
 107
 108         if (p->upstream_eof || p->upstream_error || p->upstream_done) {
 109             break;
 110         }
 111
 112         if (p->preread_bufs == NULL && !p->upstream->read->ready) {
 113             break;
 114         }
 115
 116         if (p->preread_bufs) {
 117
 118             /* use the pre-read bufs if they exist */
 119
 120             chain = p->preread_bufs;
 121             p->preread_bufs = NULL;
 122             n = p->preread_size;
 123
 124             ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
 125                            "pipe preread: %z", n);
 126
 127             if (n) {
 128                 p->read = 1;
 129             }
 130
 131         } else {
 132
 133 #if (NGX_HAVE_KQUEUE)
 134
 135             /*
 136              * kqueue notifies about the end of file or a pending error.
 137              * This test allows not to allocate a buf on these conditions
 138              * and not to call c->recv_chain().
 139              */
 140
 141             if (p->upstream->read->available == 0
 142                 && p->upstream->read->pending_eof)
 143             {
 144                 p->upstream->read->ready = 0;
 145                 p->upstream->read->eof = 0;
 146                 p->upstream_eof = 1;
 147                 p->read = 1;
 148
 149                 if (p->upstream->read->kq_errno) {
 150                     p->upstream->read->error = 1;
 151                     p->upstream_error = 1;
 152                     p->upstream_eof = 0;
 153
 154                     ngx_log_error(NGX_LOG_ERR, p->log,
 155                                   p->upstream->read->kq_errno,
 156                                   "kevent() reported that upstream "
 157                                   "closed connection");
 158                 }
 159
 160                 break;
 161             }
 162 #endif
 163
 164             if (p->free_raw_bufs) {
 165
 166                 /* use the free bufs if they exist */
 167
 168                 chain = p->free_raw_bufs;
 169                 if (p->single_buf) {
 170                     p->free_raw_bufs = p->free_raw_bufs->next;
 171                     chain->next = NULL;
 172                 } else {
 173                     p->free_raw_bufs = NULL;
 174                 }
 175
 176             } else if (p->allocated < p->bufs.num) {
 177
 178                 /* allocate a new buf if it's still allowed */
 179
 180                 b = ngx_create_temp_buf(p->pool, p->bufs.size);
 181                 if (b == NULL) {
 182                     return NGX_ABORT;
 183                 }
 184
 185                 p->allocated++;
 186
 187                 chain = ngx_alloc_chain_link(p->pool);
 188                 if (chain == NULL) {
 189                     return NGX_ABORT;
 190                 }
 191
 192                 chain->buf = b;
 193                 chain->next = NULL;
 194
 195             } else if (!p->cacheable
 196                        && p->downstream->data == p->output_ctx
 197                        && p->downstream->write->ready
 198                        && !p->downstream->write->delayed)
 199             {
 200                 /*
 201                  * if the bufs are not needed to be saved in a cache and
 202                  * a downstream is ready then write the bufs to a downstream
 203                  */
 204
 205                 p->upstream_blocked = 1;
 206
 207                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
 208                                "pipe downstream ready");
 209
 210                 break;
 211
 212             } else if (p->cacheable
 213                        || p->temp_file->offset < p->max_temp_file_size)
 214             {
 215
 216                 /*
 217                  * if it is allowed, then save some bufs from r->in
 218                  * to a temporary file, and add them to a r->out chain
 219                  */
 220
 221                 rc = ngx_event_pipe_write_chain_to_temp_file(p);
 222
 223                 ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
 224                                "pipe temp offset: %O", p->temp_file->offset);
 225
 226                 if (rc == NGX_BUSY) {
 227                     break;
 228                 }
 229
 230                 if (rc == NGX_AGAIN) {
 231                     if (ngx_event_flags & NGX_USE_LEVEL_EVENT
 232                         && p->upstream->read->active
 233                         && p->upstream->read->ready)
 234                     {
 235                         if (ngx_del_event(p->upstream->read, NGX_READ_EVENT, 0)
 236                             == NGX_ERROR)
 237                         {
 238                             return NGX_ABORT;
 239                         }
 240                     }
 241                 }
 242
 243                 if (rc != NGX_OK) {
 244                     return rc;
 245                 }
 246
 247                 chain = p->free_raw_bufs;
 248                 if (p->single_buf) {
 249                     p->free_raw_bufs = p->free_raw_bufs->next;
 250                     chain->next = NULL;
 251                 } else {
 252                     p->free_raw_bufs = NULL;
 253                 }
 254
 255             } else {
 256
 257                 /* there are no bufs to read in */
 258
 259                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
 260                                "no pipe bufs to read in");
 261
 262                 break;
 263             }
 264
 265             n = p->upstream->recv_chain(p->upstream, chain);
 266
 267             ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
 268                            "pipe recv chain: %z", n);
 269
 270             if (p->free_raw_bufs) {
 271                 chain->next = p->free_raw_bufs;
 272             }
 273             p->free_raw_bufs = chain;
 274
 275             if (n == NGX_ERROR) {
 276                 p->upstream_error = 1;
 277                 return NGX_ERROR;
 278             }
 279
 280             if (n == NGX_AGAIN) {
 281                 if (p->single_buf) {
 282                     ngx_event_pipe_remove_shadow_links(chain->buf);
 283                 }
 284
 285                 break;
 286             }
 287
 288             p->read = 1;
 289
 290             if (n == 0) {
 291                 p->upstream_eof = 1;
 292                 break;
 293             }
 294         }
 295
 296         p->read_length += n;
 297         cl = chain;
 298         p->free_raw_bufs = NULL;
 299
 300         while (cl && n > 0) {
 301
 302             ngx_event_pipe_remove_shadow_links(cl->buf);
 303
 304             size = cl->buf->end - cl->buf->last;
 305
 306             if (n >= size) {
 307                 cl->buf->last = cl->buf->end;
 308
 309                 /* STUB */ cl->buf->num = p->num++;
 310
 311                 if (p->input_filter(p, cl->buf) == NGX_ERROR) {
 312                     return NGX_ABORT;
 313                 }
 314
 315                 n -= size;
 316                 ln = cl;
 317                 cl = cl->next;
 318                 ngx_free_chain(p->pool, ln);
 319
 320             } else {
 321                 cl->buf->last += n;
 322                 n = 0;
 323             }
 324         }
 325
 326         if (cl) {
 327             for (ln = cl; ln->next; ln = ln->next) { /* void */ }
 328
 329             ln->next = p->free_raw_bufs;
 330             p->free_raw_bufs = cl;
 331         }
 332     }
 333
 334 #if (NGX_DEBUG)
 335
 336     for (cl = p->busy; cl; cl = cl->next) {
 337         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
 338                        "pipe buf busy s:%d t:%d f:%d "
 339                        "%p, pos %p, size: %z "
 340                        "file: %O, size: %z",
 341                        (cl->buf->shadow ? 1 : 0),
 342                        cl->buf->temporary, cl->buf->in_file,
 343                        cl->buf->start, cl->buf->pos,
 344                        cl->buf->last - cl->buf->pos,
 345                        cl->buf->file_pos,
 346                        cl->buf->file_last - cl->buf->file_pos);
 347     }
 348
 349     for (cl = p->out; cl; cl = cl->next) {
 350         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
 351                        "pipe buf out  s:%d t:%d f:%d "
 352                        "%p, pos %p, size: %z "
 353                        "file: %O, size: %z",
 354                        (cl->buf->shadow ? 1 : 0),
 355                        cl->buf->temporary, cl->buf->in_file,
 356                        cl->buf->start, cl->buf->pos,
 357                        cl->buf->last - cl->buf->pos,
 358                        cl->buf->file_pos,
 359                        cl->buf->file_last - cl->buf->file_pos);
 360     }
 361
 362     for (cl = p->in; cl; cl = cl->next) {
 363         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
 364                        "pipe buf in   s:%d t:%d f:%d "
 365                        "%p, pos %p, size: %z "
 366                        "file: %O, size: %z",
 367                        (cl->buf->shadow ? 1 : 0),
 368                        cl->buf->temporary, cl->buf->in_file,
 369                        cl->buf->start, cl->buf->pos,
 370                        cl->buf->last - cl->buf->pos,
 371                        cl->buf->file_pos,
 372                        cl->buf->file_last - cl->buf->file_pos);
 373     }
 374
 375     for (cl = p->free_raw_bufs; cl; cl = cl->next) {
 376         ngx_log_debug8(NGX_LOG_DEBUG_EVENT, p->log, 0,
 377                        "pipe buf free 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     }
 387
 388 #endif
 389
 390     if ((p->upstream_eof || p->upstream_error) && p->free_raw_bufs) {
 391
 392         /* STUB */ p->free_raw_bufs->buf->num = p->num++;
 393
 394         if (p->input_filter(p, p->free_raw_bufs->buf) == NGX_ERROR) {
 395             return NGX_ABORT;
 396         }
 397
 398         p->free_raw_bufs = p->free_raw_bufs->next;
 399
 400         if (p->free_bufs) {
 401             for (cl = p->free_raw_bufs; cl; cl = cl->next) {
 402                 if (cl->buf->shadow == NULL) {
 403                     ngx_pfree(p->pool, cl->buf->start);
 404                 }
 405             }
 406         }
 407     }
 408
 409     if (p->cacheable && p->in) {
 410         if (ngx_event_pipe_write_chain_to_temp_file(p) == NGX_ABORT) {
 411             return NGX_ABORT;
 412         }
 413     }
 414
 415     return NGX_OK;
 416 }
 417
 418
 419 static ngx_int_t
 420 ngx_event_pipe_write_to_downstream(ngx_event_pipe_t *p)
 421 {
 422     u_char            *prev;
 423     size_t             bsize;
 424     ngx_int_t          rc;
 425     ngx_uint_t         flush, prev_last_shadow;
 426     ngx_chain_t       *out, **ll, *cl;
 427     ngx_connection_t  *downstream;
 428
 429     downstream = p->downstream;
 430
 431     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
 432                    "pipe write downstream: %d", downstream->write->ready);
 433
 434     for ( ;; ) {
 435         if (p->downstream_error) {
 436             return ngx_event_pipe_drain_chains(p);
 437         }
 438
 439         if (p->upstream_eof || p->upstream_error || p->upstream_done) {
 440
 441             /* pass the p->out and p->in chains to the output filter */
 442
 443             for (cl = p->busy; cl; cl = cl->next) {
 444                 cl->buf->recycled = 0;
 445             }
 446
 447             if (p->out) {
 448                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
 449                                "pipe write downstream flush out");
 450
 451                 for (cl = p->out; cl; cl = cl->next) {
 452                     cl->buf->recycled = 0;
 453                 }
 454
 455                 rc = p->output_filter(p->output_ctx, p->out);
 456
 457                 if (downstream->destroyed) {
 458                     return NGX_ABORT;
 459                 }
 460
 461                 if (rc == NGX_ERROR) {
 462                     p->downstream_error = 1;
 463                     return ngx_event_pipe_drain_chains(p);
 464                 }
 465
 466                 p->out = NULL;
 467             }
 468
 469             if (p->in) {
 470                 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
 471                                "pipe write downstream flush in");
 472
 473                 for (cl = p->in; cl; cl = cl->next) {
 474                     cl->buf->recycled = 0;
 475                 }
 476
 477                 rc = p->output_filter(p->output_ctx, p->in);
 478
 479                 if (downstream->destroyed) {
 480                     return NGX_ABORT;
 481                 }
 482
 483                 if (rc == NGX_ERROR) {
 484                     p->downstream_error = 1;
 485                     return ngx_event_pipe_drain_chains(p);
 486                 }
 487
 488                 p->in = NULL;
 489             }
 490
 491             ngx_log_debug0(NGX_LOG_DEBUG_EVENT, p->log, 0,
 492                            "pipe write downstream done");
 493
 494             /* TODO: free unused bufs */
 495
 496             p->downstream_done = 1;
 497             break;
 498         }
 499
 500         if (downstream->data != p->output_ctx
 501             || !downstream->write->ready
 502             || downstream->write->delayed)
 503         {
 504             break;
 505         }
 506
 507         /* bsize is the size of the busy recycled bufs */
 508
 509         prev = NULL;
 510         bsize = 0;
 511
 512         for (cl = p->busy; cl; cl = cl->next) {
 513
 514             if (cl->buf->recycled) {
 515                 if (prev == cl->buf->start) {
 516                     continue;
 517                 }
 518
 519                 bsize += cl->buf->end - cl->buf->start;
 520                 prev = cl->buf->start;
 521             }
 522         }
 523
 524         ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
 525                        "pipe write busy: %uz", bsize);
 526
 527         out = NULL;
 528
 529         if (bsize >= (size_t) p->busy_size) {
 530             flush = 1;
 531             goto flush;
 532         }
 533
 534         flush = 0;
 535         ll = NULL;
 536         prev_last_shadow = 1;
 537
 538         for ( ;; ) {
 539             if (p->out) {
 540                 cl = p->out;
 541
 542                 if (cl->buf->recycled
 543                     && bsize + cl->buf->last - cl->buf->pos > p->busy_size)
 544                 {
 545                     flush = 1;
 546                     break;
 547                 }
 548
 549                 p->out = p->out->next;
 550
 551                 ngx_event_pipe_free_shadow_raw_buf(&p->free_raw_bufs, cl->buf);
 552
 553             } else if (!p->cacheable && p->in) {
 554                 cl = p->in;
 555
 556                 ngx_log_debug3(NGX_LOG_DEBUG_EVENT, p->log, 0,
 557                                "pipe write buf ls:%d %p %z",
 558                                cl->buf->last_shadow,
 559                                cl->buf->pos,
 560                                cl->buf->last - cl->buf->pos);
 561
 562                 if (cl->buf->recycled
 563                     && cl->buf->last_shadow
 564                     && bsize + cl->buf->last - cl->buf->pos > p->busy_size)
 565                 {
 566                     if (!prev_last_shadow) {
 567                         p->in = p->in->next;
 568
 569                         cl->next = NULL;
 570
 571                         if (out) {
 572                             *ll = cl;
 573                         } else {
 574                             out = cl;
 575                         }
 576                     }
 577
 578                     flush = 1;
 579                     break;
 580                 }
 581
 582                 prev_last_shadow = cl->buf->last_shadow;
 583
 584                 p->in = p->in->next;
 585
 586             } else {
 587                 break;
 588             }
 589
 590             if (cl->buf->recycled) {
 591                 bsize += cl->buf->last - cl->buf->pos;
 592             }
 593
 594             cl->next = NULL;
 595
 596             if (out) {
 597                 *ll = cl;
 598             } else {
 599                 out = cl;
 600             }
 601             ll = &cl->next;
 602         }
 603
 604     flush:
 605
 606         ngx_log_debug2(NGX_LOG_DEBUG_EVENT, p->log, 0,
 607                        "pipe write: out:%p, f:%d", out, flush);
 608
 609         if (out == NULL && !flush) {
 610             break;
 611         }
 612
 613         rc = p->output_filter(p->output_ctx, out);
 614
 615         if (downstream->destroyed) {
 616             return NGX_ABORT;
 617         }
 618
 619         if (rc == NGX_ERROR) {
 620             p->downstream_error = 1;
 621             return ngx_event_pipe_drain_chains(p);
 622         }
 623
 624         ngx_chain_update_chains(&p->free, &p->busy, &out, p->tag);
 625
 626         for (cl = p->free; cl; cl = cl->next) {
 627
 628             if (cl->buf->temp_file) {
 629                 if (p->cacheable || !p->cyclic_temp_file) {
 630                     continue;
 631                 }
 632
 633                 /* reset p->temp_offset if all bufs had been sent */
 634
 635                 if (cl->buf->file_last == p->temp_file->offset) {
 636                     p->temp_file->offset = 0;
 637                 }
 638             }
 639
 640             /* TODO: free buf if p->free_bufs && upstream done */
 641
 642             /* add the free shadow raw buf to p->free_raw_bufs */
 643
 644             if (cl->buf->last_shadow) {
 645                 if (ngx_event_pipe_add_free_buf(p, cl->buf->shadow) != NGX_OK) {
 646                     return NGX_ABORT;
 647                 }
 648
 649                 cl->buf->last_shadow = 0;
 650             }
 651
 652             cl->buf->shadow = NULL;
 653         }
 654     }
 655
 656     return NGX_OK;
 657 }
 658
 659
 660 static ngx_int_t
 661 ngx_event_pipe_write_chain_to_temp_file(ngx_event_pipe_t *p)
 662 {
 663     ssize_t       size, bsize;
 664     ngx_buf_t    *b;
 665     ngx_chain_t  *cl, *tl, *next, *out, **ll, **last_free, fl;
 666
 667     if (p->buf_to_file) {
 668         fl.buf = p->buf_to_file;
 669         fl.next = p->in;
 670         out = &fl;
 671
 672     } else {
 673         out = p->in;
 674     }
 675
 676     if (!p->cacheable) {
 677
 678         size = 0;
 679         cl = out;
 680         ll = NULL;
 681
 682         ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0,
 683                        "pipe offset: %O", p->temp_file->offset);
 684
 685         do {
 686             bsize = cl->buf->last - cl->buf->pos;
 687
 688             ngx_log_debug3(NGX_LOG_DEBUG_EVENT, p->log, 0,
 689                            "pipe buf %p, pos %p, size: %z",
 690                            cl->buf->start, cl->buf->pos, bsize);
 691
 692             if ((size + bsize > p->temp_file_write_size)
 693                || (p->temp_file->offset + size + bsize > p->max_temp_file_size))
 694             {
 695                 break;
 696             }
 697
 698             size += bsize;
 699             ll = &cl->next;
 700             cl = cl->next;
 701
 702         } while (cl);
 703
 704         ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0, "size: %z", size);
 705
 706         if (ll == NULL) {
 707             return NGX_BUSY;
 708         }
 709
 710         if (cl) {
 711            p->in = cl;
 712            *ll = NULL;
 713
 714         } else {
 715            p->in = NULL;
 716            p->last_in = &p->in;
 717         }
 718
 719     } else {
 720         p->in = NULL;
 721         p->last_in = &p->in;
 722     }
 723
 724     if (ngx_write_chain_to_temp_file(p->temp_file, out) == NGX_ERROR) {
 725         return NGX_ABORT;
 726     }
 727
 728     for (last_free = &p->free_raw_bufs;
 729          *last_free != NULL;
 730          last_free = &(*last_free)->next)
 731     {
 732         /* void */
 733     }
 734
 735     if (p->buf_to_file) {
 736         p->temp_file->offset = p->buf_to_file->last - p->buf_to_file->pos;
 737         p->buf_to_file = NULL;
 738         out = out->next;
 739     }
 740
 741     for (cl = out; cl; cl = next) {
 742         next = cl->next;
 743         cl->next = NULL;
 744
 745         b = cl->buf;
 746         b->file = &p->temp_file->file;
 747         b->file_pos = p->temp_file->offset;
 748         p->temp_file->offset += b->last - b->pos;
 749         b->file_last = p->temp_file->offset;
 750
 751         b->in_file = 1;
 752         b->temp_file = 1;
 753
 754         if (p->out) {
 755             *p->last_out = cl;
 756         } else {
 757             p->out = cl;
 758         }
 759         p->last_out = &cl->next;
 760
 761         if (b->last_shadow) {
 762
 763             tl = ngx_alloc_chain_link(p->pool);
 764             if (tl == NULL) {
 765                 return NGX_ABORT;
 766             }
 767
 768             tl->buf = b->shadow;
 769             tl->next = NULL;
 770
 771             *last_free = tl;
 772             last_free = &tl->next;
 773
 774             b->shadow->pos = b->shadow->start;
 775             b->shadow->last = b->shadow->start;
 776
 777             ngx_event_pipe_remove_shadow_links(b->shadow);
 778         }
 779     }
 780
 781     return NGX_OK;
 782 }
 783
 784
 785 /* the copy input filter */
 786
 787 ngx_int_t
 788 ngx_event_pipe_copy_input_filter(ngx_event_pipe_t *p, ngx_buf_t *buf)
 789 {
 790     ngx_buf_t    *b;
 791     ngx_chain_t  *cl;
 792
 793     if (buf->pos == buf->last) {
 794         return NGX_OK;
 795     }
 796
 797     if (p->free) {
 798         cl = p->free;
 799         b = cl->buf;
 800         p->free = cl->next;
 801         ngx_free_chain(p->pool, cl);
 802
 803     } else {
 804         b = ngx_alloc_buf(p->pool);
 805         if (b == NULL) {
 806             return NGX_ERROR;
 807         }
 808     }
 809
 810     ngx_memcpy(b, buf, sizeof(ngx_buf_t));
 811     b->shadow = buf;
 812     b->tag = p->tag;
 813     b->last_shadow = 1;
 814     b->recycled = 1;
 815     buf->shadow = b;
 816
 817     cl = ngx_alloc_chain_link(p->pool);
 818     if (cl == NULL) {
 819         return NGX_ERROR;
 820     }
 821
 822     cl->buf = b;
 823     cl->next = NULL;
 824
 825     ngx_log_debug1(NGX_LOG_DEBUG_EVENT, p->log, 0, "input buf #%d", b->num);
 826
 827     if (p->in) {
 828         *p->last_in = cl;
 829     } else {
 830         p->in = cl;
 831     }
 832     p->last_in = &cl->next;
 833
 834     return NGX_OK;
 835 }
 836
 837
 838 static ngx_inline void
 839 ngx_event_pipe_remove_shadow_links(ngx_buf_t *buf)
 840 {
 841     ngx_buf_t  *b, *next;
 842
 843     b = buf->shadow;
 844
 845     if (b == NULL) {
 846         return;
 847     }
 848
 849     while (!b->last_shadow) {
 850         next = b->shadow;
 851
 852         b->temporary = 0;
 853         b->recycled = 0;
 854
 855         b->shadow = NULL;
 856         b = next;
 857     }
 858
 859     b->temporary = 0;
 860     b->recycled = 0;
 861     b->last_shadow = 0;
 862
 863     b->shadow = NULL;
 864
 865     buf->shadow = NULL;
 866 }
 867
 868
 869 static ngx_inline void
 870 ngx_event_pipe_free_shadow_raw_buf(ngx_chain_t **free, ngx_buf_t *buf)
 871 {
 872     ngx_buf_t    *s;
 873     ngx_chain_t  *cl, **ll;
 874
 875     if (buf->shadow == NULL) {
 876         return;
 877     }
 878
 879     for (s = buf->shadow; !s->last_shadow; s = s->shadow) { /* void */ }
 880
 881     ll = free;
 882
 883     for (cl = *free; cl; cl = cl->next) {
 884         if (cl->buf == s) {
 885             *ll = cl->next;
 886             break;
 887         }
 888
 889         if (cl->buf->shadow) {
 890             break;
 891         }
 892
 893         ll = &cl->next;
 894     }
 895 }
 896
 897
 898 ngx_int_t
 899 ngx_event_pipe_add_free_buf(ngx_event_pipe_t *p, ngx_buf_t *b)
 900 {
 901     ngx_chain_t  *cl;
 902
 903     cl = ngx_alloc_chain_link(p->pool);
 904     if (cl == NULL) {
 905         return NGX_ERROR;
 906     }
 907
 908     b->pos = b->start;
 909     b->last = b->start;
 910     b->shadow = NULL;
 911
 912     cl->buf = b;
 913
 914     if (p->free_raw_bufs == NULL) {
 915         p->free_raw_bufs = cl;
 916         cl->next = NULL;
 917
 918         return NGX_OK;
 919     }
 920
 921     if (p->free_raw_bufs->buf->pos == p->free_raw_bufs->buf->last) {
 922
 923         /* add the free buf to the list start */
 924
 925         cl->next = p->free_raw_bufs;
 926         p->free_raw_bufs = cl;
 927
 928         return NGX_OK;
 929     }
 930
 931     /* the first free buf is partialy filled, thus add the free buf after it */
 932
 933     cl->next = p->free_raw_bufs->next;
 934     p->free_raw_bufs->next = cl;
 935
 936     return NGX_OK;
 937 }
 938
 939
 940 static ngx_int_t
 941 ngx_event_pipe_drain_chains(ngx_event_pipe_t *p)
 942 {
 943     ngx_chain_t  *cl, *tl;
 944
 945     for ( ;; ) {
 946         if (p->busy) {
 947             cl = p->busy;
 948             p->busy = NULL;
 949
 950         } else if (p->out) {
 951             cl = p->out;
 952             p->out = NULL;
 953
 954         } else if (p->in) {
 955             cl = p->in;
 956             p->in = NULL;
 957
 958         } else {
 959             return NGX_OK;
 960         }
 961
 962         while (cl) {
 963             if (cl->buf->last_shadow) {
 964                 if (ngx_event_pipe_add_free_buf(p, cl->buf->shadow) != NGX_OK) {
 965                     return NGX_ABORT;
 966                 }
 967
 968                 cl->buf->last_shadow = 0;
 969             }
 970
 971             cl->buf->shadow = NULL;
 972             tl = cl->next;
 973             cl->next = p->free;
 974             p->free = cl;
 975             cl = tl;
 976         }
 977     }
 978 }