src/chunk.c

   1 #include "first.h"
   2
   3 /**
   4  * the network chunk-API
   5  *
   6  *
   7  */
   8
   9 #include "chunk.h"
  10 #include "base.h"
  11 #include "log.h"
  12
  13 #include <sys/types.h>
  14 #include <sys/stat.h>
  15 #include "sys-mmap.h"
  16
  17 #include <stdlib.h>
  18 #include <fcntl.h>
  19 #include <unistd.h>
  20
  21 #include <stdio.h>
  22 #include <errno.h>
  23 #include <string.h>
  24
  25 chunkqueue *chunkqueue_init(void) {
  26         chunkqueue *cq;
  27
  28         cq = calloc(1, sizeof(*cq));
  29         force_assert(NULL != cq);
  30
  31         cq->first = NULL;
  32         cq->last = NULL;
  33
  34         cq->unused = NULL;
  35
  36         return cq;
  37 }
  38
  39 static chunk *chunk_init(void) {
  40         chunk *c;
  41
  42         c = calloc(1, sizeof(*c));
  43         force_assert(NULL != c);
  44
  45         c->type = MEM_CHUNK;
  46         c->mem = buffer_init();
  47         c->file.name = buffer_init();
  48         c->file.start = c->file.length = c->file.mmap.offset = 0;
  49         c->file.fd = -1;
  50         c->file.mmap.start = MAP_FAILED;
  51         c->file.mmap.length = 0;
  52         c->file.is_temp = 0;
  53         c->offset = 0;
  54         c->next = NULL;
  55
  56         return c;
  57 }
  58
  59 static void chunk_reset(chunk *c) {
  60         if (NULL == c) return;
  61
  62         c->type = MEM_CHUNK;
  63
  64         buffer_reset(c->mem);
  65
  66         if (c->file.is_temp && !buffer_string_is_empty(c->file.name)) {
  67                 unlink(c->file.name->ptr);
  68         }
  69
  70         buffer_reset(c->file.name);
  71
  72         if (c->file.fd != -1) {
  73                 close(c->file.fd);
  74                 c->file.fd = -1;
  75         }
  76         if (MAP_FAILED != c->file.mmap.start) {
  77                 munmap(c->file.mmap.start, c->file.mmap.length);
  78                 c->file.mmap.start = MAP_FAILED;
  79         }
  80         c->file.start = c->file.length = c->file.mmap.offset = 0;
  81         c->file.mmap.length = 0;
  82         c->file.is_temp = 0;
  83         c->offset = 0;
  84         c->next = NULL;
  85 }
  86
  87 static void chunk_free(chunk *c) {
  88         if (NULL == c) return;
  89
  90         chunk_reset(c);
  91
  92         buffer_free(c->mem);
  93         buffer_free(c->file.name);
  94
  95         free(c);
  96 }
  97
  98 static off_t chunk_remaining_length(const chunk *c) {
  99         off_t len = 0;
 100         switch (c->type) {
 101         case MEM_CHUNK:
 102                 len = buffer_string_length(c->mem);
 103                 break;
 104         case FILE_CHUNK:
 105                 len = c->file.length;
 106                 break;
 107         default:
 108                 force_assert(c->type == MEM_CHUNK || c->type == FILE_CHUNK);
 109                 break;
 110         }
 111         force_assert(c->offset <= len);
 112         return len - c->offset;
 113 }
 114
 115 void chunkqueue_free(chunkqueue *cq) {
 116         chunk *c, *pc;
 117
 118         if (NULL == cq) return;
 119
 120         for (c = cq->first; c; ) {
 121                 pc = c;
 122                 c = c->next;
 123                 chunk_free(pc);
 124         }
 125
 126         for (c = cq->unused; c; ) {
 127                 pc = c;
 128                 c = c->next;
 129                 chunk_free(pc);
 130         }
 131
 132         free(cq);
 133 }
 134
 135 static void chunkqueue_push_unused_chunk(chunkqueue *cq, chunk *c) {
 136         force_assert(NULL != cq && NULL != c);
 137
 138         /* keep at max 4 chunks in the 'unused'-cache */
 139         if (cq->unused_chunks > 4) {
 140                 chunk_free(c);
 141         } else {
 142                 chunk_reset(c);
 143                 c->next = cq->unused;
 144                 cq->unused = c;
 145                 cq->unused_chunks++;
 146         }
 147 }
 148
 149 static chunk *chunkqueue_get_unused_chunk(chunkqueue *cq) {
 150         chunk *c;
 151
 152         force_assert(NULL != cq);
 153
 154         /* check if we have a unused chunk */
 155         if (0 == cq->unused) {
 156                 c = chunk_init();
 157         } else {
 158                 /* take the first element from the list (a stack) */
 159                 c = cq->unused;
 160                 cq->unused = c->next;
 161                 c->next = NULL;
 162                 cq->unused_chunks--;
 163         }
 164
 165         return c;
 166 }
 167
 168 static void chunkqueue_prepend_chunk(chunkqueue *cq, chunk *c) {
 169         c->next = cq->first;
 170         cq->first = c;
 171
 172         if (NULL == cq->last) {
 173                 cq->last = c;
 174         }
 175         cq->bytes_in += chunk_remaining_length(c);
 176 }
 177
 178 static void chunkqueue_append_chunk(chunkqueue *cq, chunk *c) {
 179         c->next = NULL;
 180         if (cq->last) {
 181                 cq->last->next = c;
 182         }
 183         cq->last = c;
 184
 185         if (NULL == cq->first) {
 186                 cq->first = c;
 187         }
 188         cq->bytes_in += chunk_remaining_length(c);
 189 }
 190
 191 void chunkqueue_reset(chunkqueue *cq) {
 192         chunk *cur = cq->first;
 193
 194         cq->first = cq->last = NULL;
 195
 196         while (NULL != cur) {
 197                 chunk *next = cur->next;
 198                 chunkqueue_push_unused_chunk(cq, cur);
 199                 cur = next;
 200         }
 201
 202         cq->bytes_in = 0;
 203         cq->bytes_out = 0;
 204         cq->tempdir_idx = 0;
 205 }
 206
 207 void chunkqueue_append_file_fd(chunkqueue *cq, buffer *fn, int fd, off_t offset, off_t len) {
 208         chunk *c;
 209
 210         if (0 == len) {
 211                 close(fd);
 212                 return;
 213         }
 214
 215         c = chunkqueue_get_unused_chunk(cq);
 216
 217         c->type = FILE_CHUNK;
 218
 219         buffer_copy_buffer(c->file.name, fn);
 220         c->file.start = offset;
 221         c->file.length = len;
 222         c->file.fd = fd;
 223         c->offset = 0;
 224
 225         chunkqueue_append_chunk(cq, c);
 226 }
 227
 228 void chunkqueue_append_file(chunkqueue *cq, buffer *fn, off_t offset, off_t len) {
 229         chunk *c;
 230
 231         if (0 == len) return;
 232
 233         c = chunkqueue_get_unused_chunk(cq);
 234
 235         c->type = FILE_CHUNK;
 236
 237         buffer_copy_buffer(c->file.name, fn);
 238         c->file.start = offset;
 239         c->file.length = len;
 240         c->offset = 0;
 241
 242         chunkqueue_append_chunk(cq, c);
 243 }
 244
 245 void chunkqueue_append_buffer(chunkqueue *cq, buffer *mem) {
 246         chunk *c;
 247
 248         if (buffer_string_is_empty(mem)) return;
 249
 250         c = chunkqueue_get_unused_chunk(cq);
 251         c->type = MEM_CHUNK;
 252         force_assert(NULL != c->mem);
 253         buffer_move(c->mem, mem);
 254
 255         chunkqueue_append_chunk(cq, c);
 256 }
 257
 258 void chunkqueue_prepend_buffer(chunkqueue *cq, buffer *mem) {
 259         chunk *c;
 260
 261         if (buffer_string_is_empty(mem)) return;
 262
 263         c = chunkqueue_get_unused_chunk(cq);
 264         c->type = MEM_CHUNK;
 265         force_assert(NULL != c->mem);
 266         buffer_move(c->mem, mem);
 267
 268         chunkqueue_prepend_chunk(cq, c);
 269 }
 270
 271
 272 void chunkqueue_append_mem(chunkqueue *cq, const char * mem, size_t len) {
 273         chunk *c;
 274
 275         if (0 == len) return;
 276
 277         c = chunkqueue_get_unused_chunk(cq);
 278         c->type = MEM_CHUNK;
 279         buffer_copy_string_len(c->mem, mem, len);
 280
 281         chunkqueue_append_chunk(cq, c);
 282 }
 283
 284 void chunkqueue_get_memory(chunkqueue *cq, char **mem, size_t *len, size_t min_size, size_t alloc_size) {
 285         static const size_t REALLOC_MAX_SIZE = 256;
 286         chunk *c;
 287         buffer *b;
 288         char *dummy_mem;
 289         size_t dummy_len;
 290
 291         force_assert(NULL != cq);
 292         if (NULL == mem) mem = &dummy_mem;
 293         if (NULL == len) len = &dummy_len;
 294
 295         /* default values: */
 296         if (0 == min_size) min_size = 1024;
 297         if (0 == alloc_size) alloc_size = 4096;
 298         if (alloc_size < min_size) alloc_size = min_size;
 299
 300         if (NULL != cq->last && MEM_CHUNK == cq->last->type) {
 301                 size_t have;
 302
 303                 b = cq->last->mem;
 304                 have = buffer_string_space(b);
 305
 306                 /* unused buffer: allocate space */
 307                 if (buffer_string_is_empty(b)) {
 308                         buffer_string_prepare_copy(b, alloc_size);
 309                         have = buffer_string_space(b);
 310                 }
 311                 /* if buffer is really small just make it bigger */
 312                 else if (have < min_size && b->size <= REALLOC_MAX_SIZE) {
 313                         size_t cur_len = buffer_string_length(b);
 314                         size_t new_size = cur_len + min_size, append;
 315                         if (new_size < alloc_size) new_size = alloc_size;
 316
 317                         append = new_size - cur_len;
 318                         if (append >= min_size) {
 319                                 buffer_string_prepare_append(b, append);
 320                                 have = buffer_string_space(b);
 321                         }
 322                 }
 323
 324                 /* return pointer into existing buffer if large enough */
 325                 if (have >= min_size) {
 326                         *mem = b->ptr + buffer_string_length(b);
 327                         *len = have;
 328                         return;
 329                 }
 330         }
 331
 332         /* allocate new chunk */
 333         c = chunkqueue_get_unused_chunk(cq);
 334         c->type = MEM_CHUNK;
 335         chunkqueue_append_chunk(cq, c);
 336
 337         b = c->mem;
 338         buffer_string_prepare_append(b, alloc_size);
 339
 340         *mem = b->ptr + buffer_string_length(b);
 341         *len = buffer_string_space(b);
 342 }
 343
 344 void chunkqueue_use_memory(chunkqueue *cq, size_t len) {
 345         buffer *b;
 346
 347         force_assert(NULL != cq);
 348         force_assert(NULL != cq->last && MEM_CHUNK == cq->last->type);
 349         b = cq->last->mem;
 350
 351         if (len > 0) {
 352                 buffer_commit(b, len);
 353                 cq->bytes_in += len;
 354         } else if (buffer_string_is_empty(b)) {
 355                 /* unused buffer: can't remove chunk easily from
 356                  * end of list, so just reset the buffer
 357                  */
 358                 buffer_reset(b);
 359         }
 360 }
 361
 362 /* default 1MB, upper limit 128MB */
 363 #define DEFAULT_TEMPFILE_SIZE (1 * 1024 * 1024)
 364 #define MAX_TEMPFILE_SIZE (128 * 1024 * 1024)
 365
 366 void chunkqueue_set_tempdirs(chunkqueue *cq, array *tempdirs, unsigned int upload_temp_file_size) {
 367         force_assert(NULL != cq);
 368         cq->tempdirs = tempdirs;
 369         cq->upload_temp_file_size
 370                 = (0 == upload_temp_file_size)                ? DEFAULT_TEMPFILE_SIZE
 371                 : (upload_temp_file_size > MAX_TEMPFILE_SIZE) ? MAX_TEMPFILE_SIZE
 372                                                               : upload_temp_file_size;
 373         cq->tempdir_idx = 0;
 374 }
 375
 376 void chunkqueue_steal(chunkqueue *dest, chunkqueue *src, off_t len) {
 377         while (len > 0) {
 378                 chunk *c = src->first;
 379                 off_t clen = 0, use;
 380
 381                 if (NULL == c) break;
 382
 383                 clen = chunk_remaining_length(c);
 384                 if (0 == clen) {
 385                         /* drop empty chunk */
 386                         src->first = c->next;
 387                         if (c == src->last) src->last = NULL;
 388                         chunkqueue_push_unused_chunk(src, c);
 389                         continue;
 390                 }
 391
 392                 use = len >= clen ? clen : len;
 393                 len -= use;
 394
 395                 if (use == clen) {
 396                         /* move complete chunk */
 397                         src->first = c->next;
 398                         if (c == src->last) src->last = NULL;
 399
 400                         chunkqueue_append_chunk(dest, c);
 401                 } else {
 402                         /* partial chunk with length "use" */
 403
 404                         switch (c->type) {
 405                         case MEM_CHUNK:
 406                                 chunkqueue_append_mem(dest, c->mem->ptr + c->offset, use);
 407                                 break;
 408                         case FILE_CHUNK:
 409                                 /* tempfile flag is in "last" chunk after the split */
 410                                 chunkqueue_append_file(dest, c->file.name, c->file.start + c->offset, use);
 411                                 break;
 412                         }
 413
 414                         c->offset += use;
 415                         force_assert(0 == len);
 416                 }
 417
 418                 src->bytes_out += use;
 419         }
 420 }
 421
 422 static chunk *chunkqueue_get_append_tempfile(chunkqueue *cq) {
 423         chunk *c;
 424         buffer *template = buffer_init_string("/var/tmp/lighttpd-upload-XXXXXX");
 425         int fd;
 426
 427         if (cq->tempdirs && cq->tempdirs->used) {
 428                 /* we have several tempdirs, only if all of them fail we jump out */
 429
 430                 for (errno = EIO; cq->tempdir_idx < cq->tempdirs->used; ++cq->tempdir_idx) {
 431                         data_string *ds = (data_string *)cq->tempdirs->data[cq->tempdir_idx];
 432
 433                         buffer_copy_buffer(template, ds->value);
 434                         buffer_append_slash(template);
 435                         buffer_append_string_len(template, CONST_STR_LEN("lighttpd-upload-XXXXXX"));
 436
 437                         if (-1 != (fd = mkstemp(template->ptr))) break;
 438                 }
 439         } else {
 440                 fd = mkstemp(template->ptr);
 441         }
 442
 443         if (-1 == fd) {
 444                 buffer_free(template);
 445                 return NULL;
 446         }
 447
 448         c = chunkqueue_get_unused_chunk(cq);
 449         c->type = FILE_CHUNK;
 450         c->file.fd = fd;
 451         c->file.is_temp = 1;
 452         buffer_copy_buffer(c->file.name, template);
 453         c->file.length = 0;
 454
 455         chunkqueue_append_chunk(cq, c);
 456
 457         buffer_free(template);
 458
 459         return c;
 460 }
 461
 462 static void chunkqueue_remove_empty_chunks(chunkqueue *cq);
 463
 464 static int chunkqueue_append_to_tempfile(server *srv, chunkqueue *dest, const char *mem, size_t len) {
 465         chunk *dst_c;
 466         ssize_t written;
 467
 468         do {
 469                 /*
 470                  * if the last chunk is
 471                  * - smaller than dest->upload_temp_file_size
 472                  * - not read yet (offset == 0)
 473                  * -> append to it (so it might actually become larger than dest->upload_temp_file_size)
 474                  * otherwise
 475                  * -> create a new chunk
 476                  *
 477                  * */
 478
 479                 dst_c = dest->last;
 480                 if (NULL != dst_c
 481                         && FILE_CHUNK == dst_c->type
 482                         && dst_c->file.is_temp
 483                         && -1 != dst_c->file.fd
 484                         && 0 == dst_c->offset) {
 485                         /* ok, take the last chunk for our job */
 486
 487                         if (dst_c->file.length >= (off_t)dest->upload_temp_file_size) {
 488                                 /* the chunk is too large now, close it */
 489                                 int rc = close(dst_c->file.fd);
 490                                 dst_c->file.fd = -1;
 491                                 dst_c = NULL;
 492                                 if (0 != rc) {
 493                                         log_error_write(srv, __FILE__, __LINE__, "sbss",
 494                                                 "close() temp-file", dst_c->file.name, "failed:",
 495                                                 strerror(errno));
 496                                         return -1;
 497                                 }
 498                         }
 499                 } else {
 500                         dst_c = NULL;
 501                 }
 502
 503                 if (NULL == dst_c && NULL == (dst_c = chunkqueue_get_append_tempfile(dest))) {
 504                         /* we don't have file to write to,
 505                          * EACCES might be one reason.
 506                          *
 507                          * Instead of sending 500 we send 413 and say the request is too large
 508                          */
 509
 510                         log_error_write(srv, __FILE__, __LINE__, "ss",
 511                                 "opening temp-file failed:", strerror(errno));
 512
 513                         return -1;
 514                 }
 515
 516                 written = write(dst_c->file.fd, mem, len);
 517
 518                 if ((size_t) written == len) {
 519                         dst_c->file.length += len;
 520                         dest->bytes_in += len;
 521
 522                         return 0;
 523                 } else if (written >= 0) {
 524                         /*(assume EINTR if partial write and retry write();
 525                          * retry write() might fail with ENOSPC if no more space on volume)*/
 526                         dest->bytes_in += written;
 527                         mem += written;
 528                         len -= (size_t)written;
 529                         dst_c->file.length += (size_t)written;
 530                         /* continue; retry */
 531                 } else if (errno == EINTR) {
 532                         /* continue; retry */
 533                 } else {
 534                         int retry = (errno == ENOSPC && dest->tempdirs && ++dest->tempdir_idx < dest->tempdirs->used);
 535                         if (!retry) {
 536                                 log_error_write(srv, __FILE__, __LINE__, "sbs",
 537                                                 "write() temp-file", dst_c->file.name, "failed:",
 538                                                 strerror(errno));
 539                         }
 540
 541                         if (0 == chunk_remaining_length(dst_c)) {
 542                                 /*(remove empty chunk and unlink tempfile)*/
 543                                 chunkqueue_remove_empty_chunks(dest);
 544                         } else {/*(close tempfile; avoid later attempts to append)*/
 545                                 int rc = close(dst_c->file.fd);
 546                                 dst_c->file.fd = -1;
 547                                 if (0 != rc) {
 548                                         log_error_write(srv, __FILE__, __LINE__, "sbss",
 549                                                 "close() temp-file", dst_c->file.name, "failed:",
 550                                                 strerror(errno));
 551                                         return -1;
 552                                 }
 553                         }
 554                         if (!retry) return -1;
 555
 556                         /* continue; retry */
 557                 }
 558
 559         } while (dst_c);
 560
 561         return -1; /*(not reached)*/
 562 }
 563
 564 int chunkqueue_steal_with_tempfiles(server *srv, chunkqueue *dest, chunkqueue *src, off_t len) {
 565         while (len > 0) {
 566                 chunk *c = src->first;
 567                 off_t clen = 0, use;
 568
 569                 if (NULL == c) break;
 570
 571                 clen = chunk_remaining_length(c);
 572                 if (0 == clen) {
 573                         /* drop empty chunk */
 574                         src->first = c->next;
 575                         if (c == src->last) src->last = NULL;
 576                         chunkqueue_push_unused_chunk(src, c);
 577                         continue;
 578                 }
 579
 580                 use = (len >= clen) ? clen : len;
 581                 len -= use;
 582
 583                 switch (c->type) {
 584                 case FILE_CHUNK:
 585                         if (use == clen) {
 586                                 /* move complete chunk */
 587                                 src->first = c->next;
 588                                 if (c == src->last) src->last = NULL;
 589                                 chunkqueue_append_chunk(dest, c);
 590                         } else {
 591                                 /* partial chunk with length "use" */
 592                                 /* tempfile flag is in "last" chunk after the split */
 593                                 chunkqueue_append_file(dest, c->file.name, c->file.start + c->offset, use);
 594
 595                                 c->offset += use;
 596                                 force_assert(0 == len);
 597                         }
 598                         break;
 599
 600                 case MEM_CHUNK:
 601                         /* store "use" bytes from memory chunk in tempfile */
 602                         if (0 != chunkqueue_append_to_tempfile(srv, dest, c->mem->ptr + c->offset, use)) {
 603                                 return -1;
 604                         }
 605
 606                         if (use == clen) {
 607                                 /* finished chunk */
 608                                 src->first = c->next;
 609                                 if (c == src->last) src->last = NULL;
 610                                 chunkqueue_push_unused_chunk(src, c);
 611                         } else {
 612                                 /* partial chunk */
 613                                 c->offset += use;
 614                                 force_assert(0 == len);
 615                         }
 616                         break;
 617                 }
 618
 619                 src->bytes_out += use;
 620         }
 621
 622         return 0;
 623 }
 624
 625 off_t chunkqueue_length(chunkqueue *cq) {
 626         off_t len = 0;
 627         chunk *c;
 628
 629         for (c = cq->first; c; c = c->next) {
 630                 len += chunk_remaining_length(c);
 631         }
 632
 633         return len;
 634 }
 635
 636 int chunkqueue_is_empty(chunkqueue *cq) {
 637         return NULL == cq->first;
 638 }
 639
 640 void chunkqueue_mark_written(chunkqueue *cq, off_t len) {
 641         off_t written = len;
 642         chunk *c;
 643         force_assert(len >= 0);
 644
 645         for (c = cq->first; NULL != c; c = cq->first) {
 646                 off_t c_len = chunk_remaining_length(c);
 647
 648                 if (0 == written && 0 != c_len) break; /* no more finished chunks */
 649
 650                 if (written >= c_len) { /* chunk got finished */
 651                         c->offset += c_len;
 652                         written -= c_len;
 653
 654                         cq->first = c->next;
 655                         if (c == cq->last) cq->last = NULL;
 656
 657                         chunkqueue_push_unused_chunk(cq, c);
 658                 } else { /* partial chunk */
 659                         c->offset += written;
 660                         written = 0;
 661                         break; /* chunk not finished */
 662                 }
 663         }
 664
 665         force_assert(0 == written);
 666         cq->bytes_out += len;
 667 }
 668
 669 void chunkqueue_remove_finished_chunks(chunkqueue *cq) {
 670         chunk *c;
 671
 672         for (c = cq->first; c; c = cq->first) {
 673                 if (0 != chunk_remaining_length(c)) break; /* not finished yet */
 674
 675                 cq->first = c->next;
 676                 if (c == cq->last) cq->last = NULL;
 677
 678                 chunkqueue_push_unused_chunk(cq, c);
 679         }
 680 }
 681
 682 static void chunkqueue_remove_empty_chunks(chunkqueue *cq) {
 683         chunk *c;
 684         chunkqueue_remove_finished_chunks(cq);
 685         if (chunkqueue_is_empty(cq)) return;
 686
 687         for (c = cq->first; c->next; c = c->next) {
 688                 if (0 == chunk_remaining_length(c->next)) {
 689                         chunk *empty = c->next;
 690                         c->next = empty->next;
 691                         if (empty == cq->last) cq->last = c;
 692
 693                         chunkqueue_push_unused_chunk(cq, empty);
 694                 }
 695         }
 696 }