93 chunkqueue *chunkqueue_init(chunkqueue *cq) {  in chunkqueue_init()  argument
95 	if (NULL == cq)  in chunkqueue_init()
96 		cq = ck_calloc(1, sizeof(*cq));  in chunkqueue_init()
99 	cq->first = NULL;  in chunkqueue_init()
100 	cq->last = NULL;  in chunkqueue_init()
103 	cq->tempdirs              = chunkqueue_default_tempdirs;  in chunkqueue_init()
104 	cq->upload_temp_file_size = chunkqueue_default_tempfile_size;  in chunkqueue_init()
106 	return cq;  in chunkqueue_init()
438 static void chunkqueue_release_chunks(chunkqueue *cq) {  in chunkqueue_release_chunks()  argument
439     cq->last = NULL;  in chunkqueue_release_chunks()
440     for (chunk *c; (c = cq->first); ) {  in chunkqueue_release_chunks()
441         cq->first = c->next;  in chunkqueue_release_chunks()
446 void chunkqueue_free(chunkqueue *cq) {  in chunkqueue_free()  argument
447     if (NULL == cq) return;  in chunkqueue_free()
448     chunkqueue_release_chunks(cq);  in chunkqueue_free()
449     free(cq);  in chunkqueue_free()
452 static void chunkqueue_prepend_chunk(chunkqueue * const restrict cq, chunk * const restrict c) {  in chunkqueue_prepend_chunk()  argument
453     if (NULL == (c->next = cq->first)) cq->last = c;  in chunkqueue_prepend_chunk()
454     cq->first = c;  in chunkqueue_prepend_chunk()
457 static void chunkqueue_append_chunk(chunkqueue * const restrict cq, chunk * const restrict c) {  in chunkqueue_append_chunk()  argument
459     *(cq->last ? &cq->last->next : &cq->first) = c;  in chunkqueue_append_chunk()
460     cq->last = c;  in chunkqueue_append_chunk()
464 static chunk * chunkqueue_prepend_mem_chunk(chunkqueue *cq, size_t sz) {  in chunkqueue_prepend_mem_chunk()  argument
466     chunkqueue_prepend_chunk(cq, c);  in chunkqueue_prepend_mem_chunk()
471 static chunk * chunkqueue_append_mem_chunk(chunkqueue *cq, size_t sz) {  in chunkqueue_append_mem_chunk()  argument
473     chunkqueue_append_chunk(cq, c);  in chunkqueue_append_mem_chunk()
479 static chunk * chunkqueue_append_file_chunk(chunkqueue * const restrict cq, const buffer * const re…  in __attribute_nonnull__()
481     chunkqueue_append_chunk(cq, c);  in __attribute_nonnull__()
485     cq->bytes_in += len;  in __attribute_nonnull__()
490 void chunkqueue_reset(chunkqueue *cq) {  in chunkqueue_reset()  argument
491     chunkqueue_release_chunks(cq);  in chunkqueue_reset()
492     cq->bytes_in = 0;  in chunkqueue_reset()
493     cq->bytes_out = 0;  in chunkqueue_reset()
494     cq->tempdir_idx = 0;  in chunkqueue_reset()
497 void chunkqueue_append_file_fd(chunkqueue * const restrict cq, const buffer * const restrict fn, in…  in chunkqueue_append_file_fd()  argument
499         (chunkqueue_append_file_chunk(cq, fn, offset, len))->file.fd = fd;  in chunkqueue_append_file_fd()
506 void chunkqueue_append_file(chunkqueue * const restrict cq, const buffer * const restrict fn, off_t…  in chunkqueue_append_file()  argument
508         chunkqueue_append_file_chunk(cq, fn, offset, len);  in chunkqueue_append_file()
513 static int chunkqueue_append_mem_extend_chunk(chunkqueue * const restrict cq, const char * const re…  in chunkqueue_append_mem_extend_chunk()  argument
514 	chunk *c = cq->last;  in chunkqueue_append_mem_extend_chunk()
519 		cq->bytes_in += len;  in chunkqueue_append_mem_extend_chunk()
526 void chunkqueue_append_buffer(chunkqueue * const restrict cq, buffer * const restrict mem) {  in chunkqueue_append_buffer()  argument
529 	if (len < 1024 && chunkqueue_append_mem_extend_chunk(cq, mem->ptr, len)) {  in chunkqueue_append_buffer()
534 	c = chunkqueue_append_mem_chunk(cq, chunk_buf_sz);  in chunkqueue_append_buffer()
535 	cq->bytes_in += len;  in chunkqueue_append_buffer()
540 void chunkqueue_append_mem(chunkqueue * const restrict cq, const char * const restrict mem, size_t …  in chunkqueue_append_mem()  argument
542 	if (len < chunk_buf_sz && chunkqueue_append_mem_extend_chunk(cq, mem, len))  in chunkqueue_append_mem()
545 	c = chunkqueue_append_mem_chunk(cq, len+1);  in chunkqueue_append_mem()
546 	cq->bytes_in += len;  in chunkqueue_append_mem()
551 void chunkqueue_append_mem_min(chunkqueue * const restrict cq, const char * const restrict mem, siz…  in chunkqueue_append_mem_min()  argument
553 	if (len < chunk_buf_sz && chunkqueue_append_mem_extend_chunk(cq, mem, len))  in chunkqueue_append_mem_min()
557 	chunkqueue_append_chunk(cq, c);  in chunkqueue_append_mem_min()
558 	cq->bytes_in += len;  in chunkqueue_append_mem_min()
563 void chunkqueue_append_chunkqueue(chunkqueue * const restrict cq, chunkqueue * const restrict src) {  in chunkqueue_append_chunkqueue()  argument
566 	if (NULL == cq->first) {  in chunkqueue_append_chunkqueue()
567 		cq->first = src->first;  in chunkqueue_append_chunkqueue()
569 		cq->last->next = src->first;  in chunkqueue_append_chunkqueue()
571 	cq->last = src->last;  in chunkqueue_append_chunkqueue()
572 	cq->bytes_in += chunkqueue_length(src);  in chunkqueue_append_chunkqueue()
580 buffer * chunkqueue_prepend_buffer_open_sz(chunkqueue *cq, size_t sz) {  in chunkqueue_prepend_buffer_open_sz()  argument
581 	chunk * const c = chunkqueue_prepend_mem_chunk(cq, sz);  in chunkqueue_prepend_buffer_open_sz()
586 buffer * chunkqueue_prepend_buffer_open(chunkqueue *cq) {  in chunkqueue_prepend_buffer_open()  argument
587 	return chunkqueue_prepend_buffer_open_sz(cq, chunk_buf_sz);  in chunkqueue_prepend_buffer_open()
591 void chunkqueue_prepend_buffer_commit(chunkqueue *cq) {  in chunkqueue_prepend_buffer_commit()  argument
592 	cq->bytes_in += buffer_clen(cq->first->mem);  in chunkqueue_prepend_buffer_commit()
596 buffer * chunkqueue_append_buffer_open_sz(chunkqueue *cq, size_t sz) {  in chunkqueue_append_buffer_open_sz()  argument
597 	chunk * const c = chunkqueue_append_mem_chunk(cq, sz);  in chunkqueue_append_buffer_open_sz()
602 buffer * chunkqueue_append_buffer_open(chunkqueue *cq) {  in chunkqueue_append_buffer_open()  argument
603 	return chunkqueue_append_buffer_open_sz(cq, chunk_buf_sz);  in chunkqueue_append_buffer_open()
607 void chunkqueue_append_buffer_commit(chunkqueue *cq) {  in chunkqueue_append_buffer_commit()  argument
608 	cq->bytes_in += buffer_clen(cq->last->mem);  in chunkqueue_append_buffer_commit()
612 char * chunkqueue_get_memory(chunkqueue * const restrict cq, size_t * const restrict len) {  in chunkqueue_get_memory()  argument
615 	chunk *c = cq->last;  in chunkqueue_get_memory()
627 	b = chunkqueue_append_buffer_open_sz(cq, sz);  in chunkqueue_get_memory()
632 void chunkqueue_use_memory(chunkqueue * const restrict cq, chunk *ckpt, size_t len) {  in chunkqueue_use_memory()  argument
633     buffer *b = cq->last->mem;  in chunkqueue_use_memory()
637         cq->bytes_in += len;  in chunkqueue_use_memory()
638         if (cq->last == ckpt || NULL == ckpt || MEM_CHUNK != ckpt->type  in chunkqueue_use_memory()
648     chunk_release(cq->last);  in chunkqueue_use_memory()
649     cq->last = ckpt;  in chunkqueue_use_memory()
650     *(ckpt ? &ckpt->next : &cq->first) = NULL;  in chunkqueue_use_memory()
653 void chunkqueue_update_file(chunkqueue * const restrict cq, chunk *c, off_t len) {  in chunkqueue_update_file()  argument
656     cq->bytes_in += len;  in chunkqueue_update_file()
658         chunkqueue_remove_empty_chunks(cq);  in chunkqueue_update_file()
668 void chunkqueue_set_tempdirs(chunkqueue * const restrict cq, const array * const restrict tempdirs,…  in chunkqueue_set_tempdirs()  argument
671     cq->tempdirs = tempdirs;  in chunkqueue_set_tempdirs()
672     cq->upload_temp_file_size = upload_temp_file_size;  in chunkqueue_set_tempdirs()
673     cq->tempdir_idx = 0;  in chunkqueue_set_tempdirs()
755 static chunk *chunkqueue_get_append_newtempfile(chunkqueue * const restrict cq, log_error_st * cons…  in chunkqueue_get_append_newtempfile()  argument
757     chunk * const restrict last = cq->last;  in chunkqueue_get_append_newtempfile()
758     chunk * const restrict c = chunkqueue_append_file_chunk(cq, &emptyb, 0, 0);  in chunkqueue_get_append_newtempfile()
762     if (cq->tempdirs && cq->tempdirs->used) {  in chunkqueue_get_append_newtempfile()
764         for (errno = EIO; cq->tempdir_idx < cq->tempdirs->used; ++cq->tempdir_idx) {  in chunkqueue_get_append_newtempfile()
765             data_string *ds = (data_string *)cq->tempdirs->data[cq->tempdir_idx];  in chunkqueue_get_append_newtempfile()
782     if ((cq->last = last))  in chunkqueue_get_append_newtempfile()
785         cq->first = NULL;  in chunkqueue_get_append_newtempfile()
790 static chunk *chunkqueue_get_append_tempfile(chunkqueue * const restrict cq, log_error_st * const r…  in chunkqueue_get_append_tempfile()  argument
799     chunk * const c = cq->last;  in chunkqueue_get_append_tempfile()
802         if (c->file.length < (off_t)cq->upload_temp_file_size)  in chunkqueue_get_append_tempfile()
815     return chunkqueue_get_append_newtempfile(cq, errh);  in chunkqueue_get_append_tempfile()
819 static int chunkqueue_append_tempfile_err(chunkqueue * const cq, log_error_st * const restrict errh…  in chunkqueue_append_tempfile_err()  argument
823     int retry = (errnum == ENOSPC && cq->tempdirs  in chunkqueue_append_tempfile_err()
824                  && ++cq->tempdir_idx < cq->tempdirs->used);  in chunkqueue_append_tempfile_err()
831         chunkqueue_remove_empty_chunks(cq);  in chunkqueue_append_tempfile_err()
1024 static ssize_t chunkqueue_append_drain_pipe_tempfile(chunkqueue * const restrict cq, const int fd, …  in chunkqueue_append_drain_pipe_tempfile()  argument
1037         if (0 != chunkqueue_append_mem_to_tempfile(cq, buf, (size_t)rd, errh))  in chunkqueue_append_drain_pipe_tempfile()
1045         if (cq->last && 0 == chunk_remaining_length(cq->last)) {  in chunkqueue_append_drain_pipe_tempfile()
1047             chunkqueue_remove_empty_chunks(cq);  in chunkqueue_append_drain_pipe_tempfile()
1053 ssize_t chunkqueue_append_splice_pipe_tempfile(chunkqueue * const restrict cq, const int fd, unsign…  in chunkqueue_append_splice_pipe_tempfile()  argument
1057     if (cq->first && cq->first->type == MEM_CHUNK) {  in chunkqueue_append_splice_pipe_tempfile()
1058         int rc = chunkqueue_to_tempfiles(cq, errh);  in chunkqueue_append_splice_pipe_tempfile()
1065         chunk * const c = chunkqueue_get_append_tempfile(cq, errh);  in chunkqueue_append_splice_pipe_tempfile()
1074             cq->bytes_in += len;  in chunkqueue_append_splice_pipe_tempfile()
1080             cq->bytes_in += wr;  in chunkqueue_append_splice_pipe_tempfile()
1097                     chunkqueue_remove_empty_chunks(cq);  in chunkqueue_append_splice_pipe_tempfile()
1101                 wr = chunkqueue_append_drain_pipe_tempfile(cq, fd, len, errh);  in chunkqueue_append_splice_pipe_tempfile()
1104                 if (!chunkqueue_append_tempfile_err(cq, errh, c))  in chunkqueue_append_splice_pipe_tempfile()
1146 ssize_t chunkqueue_append_splice_sock_tempfile(chunkqueue * const restrict cq, const int fd, unsign…  in chunkqueue_append_splice_sock_tempfile()  argument
1160     wr = chunkqueue_append_splice_pipe_tempfile(cq, pipes[0], len, errh);  in chunkqueue_append_splice_sock_tempfile()
1245 void chunkqueue_mark_written(chunkqueue *cq, off_t len) {  in chunkqueue_mark_written()  argument
1246     cq->bytes_out += len;  in chunkqueue_mark_written()
1248     for (chunk *c = cq->first; c; ) {  in chunkqueue_mark_written()
1258             cq->first = c;  in chunkqueue_mark_written()
1262     cq->first = cq->last = NULL;  in chunkqueue_mark_written()
1265 void chunkqueue_remove_finished_chunks(chunkqueue *cq) {  in chunkqueue_remove_finished_chunks()  argument
1266     for (chunk *c; (c = cq->first) && 0 == chunk_remaining_length(c); ){  in chunkqueue_remove_finished_chunks()
1267         if (NULL == (cq->first = c->next)) cq->last = NULL;  in chunkqueue_remove_finished_chunks()
1272 void chunkqueue_remove_empty_chunks(chunkqueue *cq) {  in chunkqueue_remove_empty_chunks()  argument
1274 	chunkqueue_remove_finished_chunks(cq);  in chunkqueue_remove_empty_chunks()
1276 	for (c = cq->first; c && c->next; c = c->next) {  in chunkqueue_remove_empty_chunks()
1280 			if (empty == cq->last) cq->last = c;  in chunkqueue_remove_empty_chunks()
1286 void chunkqueue_compact_mem_offset(chunkqueue * const cq) {  in chunkqueue_compact_mem_offset()  argument
1287     chunk * const restrict c = cq->first;  in chunkqueue_compact_mem_offset()
1298 void chunkqueue_compact_mem(chunkqueue *cq, size_t clen) {  in chunkqueue_compact_mem()  argument
1301     chunk *c = cq->first;  in chunkqueue_compact_mem()
1307             chunkqueue_compact_mem_offset(cq);  in chunkqueue_compact_mem()
1310         b = chunkqueue_prepend_buffer_open_sz(cq, clen+1);  in chunkqueue_compact_mem()
1312         cq->first->next = c->next;  in chunkqueue_compact_mem()
1313         if (NULL == c->next) cq->last = cq->first;  in chunkqueue_compact_mem()
1315         c = cq->first;  in chunkqueue_compact_mem()
1327         if (NULL == c->next) cq->last = fc;  in chunkqueue_compact_mem()
1361 int chunkqueue_open_file_chunk(chunkqueue * const restrict cq, log_error_st * const restrict errh) {  in chunkqueue_open_file_chunk()  argument
1362     return chunk_open_file_chunk(cq->first, errh);  in chunkqueue_open_file_chunk()
1387     chunkqueue cq = {c,c,0,0,0,0,0}; /*(fake cq for chunkqueue_peek_data())*/  in chunkqueue_write_chunk_file_intermed()  local
1388     if (0 != chunkqueue_peek_data(&cq, &data, &dlen, errh) && 0 == dlen)  in chunkqueue_write_chunk_file_intermed()
1447 chunkqueue_write_chunk (const int fd, chunkqueue * const restrict cq, log_error_st * const restrict…  in chunkqueue_write_chunk()  argument
1450     chunk * const c = cq->first;  in chunkqueue_write_chunk()
1464 chunkqueue_write_chunk_to_pipe (const int fd, chunkqueue * const restrict cq, log_error_st * const …  in chunkqueue_write_chunk_to_pipe()  argument
1467     chunk * const c = cq->first;  in chunkqueue_write_chunk_to_pipe()
1477     return chunkqueue_write_chunk(fd, cq, errh);  in chunkqueue_write_chunk_to_pipe()
1482 chunkqueue_small_resp_optim (chunkqueue * const restrict cq)  in chunkqueue_small_resp_optim()  argument
1496     chunk * restrict c = cq->first;  in chunkqueue_small_resp_optim()
1498     if (filec != cq->last || filec->type != FILE_CHUNK || filec->file.fd < 0)  in chunkqueue_small_resp_optim()
1512     cq->last = c;  in chunkqueue_small_resp_optim()
1526         cq->last = c->next = filec;  in chunkqueue_small_resp_optim()
1529         else if (__builtin_expect( (cq->first != c), 0)) {  in chunkqueue_small_resp_optim()
1530             cq->first->next = filec;  in chunkqueue_small_resp_optim()
1552 chunkqueue_peek_data (chunkqueue * const cq,  in chunkqueue_peek_data()  argument
1560     for (chunk *c = cq->first; c; ) {  in chunkqueue_peek_data()
1660 chunkqueue_read_data (chunkqueue * const cq,  in chunkqueue_read_data()  argument
1666     if (chunkqueue_peek_data(cq, &ptr, &len, errh) < 0 || len != dlen)  in chunkqueue_read_data()
1669     chunkqueue_mark_written(cq, len);  in chunkqueue_read_data()
1675 chunkqueue_read_squash (chunkqueue * const restrict cq, log_error_st * const restrict errh)  in chunkqueue_read_squash()  argument
1680     off_t cqlen = chunkqueue_length(cq);  in chunkqueue_read_squash()
1683     if (cq->first && NULL == cq->first->next && cq->first->type == MEM_CHUNK)  in chunkqueue_read_squash()
1684         return cq->first->mem;  in chunkqueue_read_squash()
1689     int rc = chunkqueue_peek_data(cq, &data, &dlen, errh);  in chunkqueue_read_squash()
1696     chunkqueue_release_chunks(cq);  in chunkqueue_read_squash()
1697     chunkqueue_append_chunk(cq, c);  in chunkqueue_read_squash()