[core] perf: chunk.c chunk pool

reuse chunks across chunkqueues; remove cq->unused, cq->unused_chunks
reuse same buffer c->mem and c->file.name in chunk

src/chunk.c

@@ -25,6 +25,8 @@
 #define DEFAULT_TEMPFILE_SIZE (1 * 1024 * 1024)
 #define MAX_TEMPFILE_SIZE (128 * 1024 * 1024)
 
+static chunk *chunks;
+static chunk *chunk_buffers;
 static array *chunkqueue_default_tempdirs = NULL;
 static unsigned int chunkqueue_default_tempfile_size = DEFAULT_TEMPFILE_SIZE;
 
@@ -43,8 +45,6 @@ chunkqueue *chunkqueue_init(void) {
 	cq->first = NULL;
 	cq->last = NULL;
 
-	cq->unused = NULL;
-
 	cq->tempdirs              = chunkqueue_default_tempdirs;
 	cq->upload_temp_file_size = chunkqueue_default_tempfile_size;
 
@@ -59,7 +59,7 @@ static chunk *chunk_init(void) {
 
 	c->type = MEM_CHUNK;
 	c->mem = buffer_init();
-	c->file.name = buffer_init();
+	c->file.name = c->mem;
 	c->file.start = c->file.length = c->file.mmap.offset = 0;
 	c->file.fd = -1;
 	c->file.mmap.start = MAP_FAILED;
@@ -68,21 +68,16 @@ static chunk *chunk_init(void) {
 	c->offset = 0;
 	c->next = NULL;
 
+	buffer_string_prepare_append(c->mem, 4095);
+
 	return c;
 }
 
 static void chunk_reset(chunk *c) {
-	if (NULL == c) return;
-
-	c->type = MEM_CHUNK;
-
-	buffer_reset(c->mem);
-
-	if (c->file.is_temp && !buffer_string_is_empty(c->file.name)) {
-		unlink(c->file.name->ptr);
+	if (c->file.is_temp && !buffer_string_is_empty(c->mem)) {
+		unlink(c->mem->ptr);
 	}
-
-	buffer_reset(c->file.name);
+	buffer_string_set_length(c->mem, 0);
 
 	if (c->file.fd != -1) {
 		close(c->file.fd);
@@ -95,21 +90,90 @@ static void chunk_reset(chunk *c) {
 	c->file.start = c->file.length = c->file.mmap.offset = 0;
 	c->file.mmap.length = 0;
 	c->file.is_temp = 0;
+	c->type = MEM_CHUNK;
 	c->offset = 0;
 	c->next = NULL;
 }
 
 static void chunk_free(chunk *c) {
-	if (NULL == c) return;
-
 	chunk_reset(c);
-
 	buffer_free(c->mem);
-	buffer_free(c->file.name);
-
 	free(c);
 }
 
+buffer * chunk_buffer_acquire(void) {
+    chunk *c;
+    buffer *b;
+    if (chunks) {
+        c = chunks;
+        chunks = c->next;
+    }
+    else {
+        c = chunk_init();
+    }
+    c->next = chunk_buffers;
+    chunk_buffers = c;
+    b = c->mem;
+    c->mem = NULL;
+    return b;
+}
+
+void chunk_buffer_release(buffer *b) {
+    if (NULL == b) return;
+    if (b->size >= 4096 && chunk_buffers) {
+        chunk *c = chunk_buffers;
+        chunk_buffers = c->next;
+        c->mem = b;
+        c->next = chunks;
+        chunks = c;
+    }
+    else {
+        buffer_free(b);
+    }
+}
+
+static chunk * chunk_acquire(void) {
+    if (chunks) {
+        chunk *c = chunks;
+        chunks = c->next;
+        return c;
+    }
+    else {
+        return chunk_init();
+    }
+}
+
+static void chunk_release(chunk *c) {
+    if (c->mem->size >= 4096) {
+        chunk_reset(c);
+        c->next = chunks;
+        chunks = c;
+    }
+    else {
+        chunk_free(c);
+    }
+}
+
+void chunkqueue_chunk_pool_clear(void)
+{
+    for (chunk *next, *c = chunks; c; c = next) {
+        next = c->next;
+        chunk_free(c);
+    }
+    chunks = NULL;
+}
+
+void chunkqueue_chunk_pool_free(void)
+{
+    chunkqueue_chunk_pool_clear();
+    for (chunk *next, *c = chunk_buffers; c; c = next) {
+        next = c->next;
+        c->mem = buffer_init(); /*(chunk_reset() expects c->mem != NULL)*/
+        chunk_free(c);
+    }
+    chunk_buffers = NULL;
+}
+
 static off_t chunk_remaining_length(const chunk *c) {
 	off_t len = 0;
 	switch (c->type) {
@@ -135,51 +199,12 @@ void chunkqueue_free(chunkqueue *cq) {
 	for (c = cq->first; c; ) {
 		pc = c;
 		c = c->next;
-		chunk_free(pc);
-	}
-
-	for (c = cq->unused; c; ) {
-		pc = c;
-		c = c->next;
-		chunk_free(pc);
+		chunk_release(pc);
 	}
 
 	free(cq);
 }
 
-static void chunkqueue_push_unused_chunk(chunkqueue *cq, chunk *c) {
-	force_assert(NULL != cq && NULL != c);
-
-	/* keep at max 4 chunks in the 'unused'-cache */
-	if (cq->unused_chunks > 4) {
-		chunk_free(c);
-	} else {
-		chunk_reset(c);
-		c->next = cq->unused;
-		cq->unused = c;
-		cq->unused_chunks++;
-	}
-}
-
-static chunk *chunkqueue_get_unused_chunk(chunkqueue *cq) {
-	chunk *c;
-
-	force_assert(NULL != cq);
-
-	/* check if we have a unused chunk */
-	if (0 == cq->unused) {
-		c = chunk_init();
-	} else {
-		/* take the first element from the list (a stack) */
-		c = cq->unused;
-		cq->unused = c->next;
-		c->next = NULL;
-		cq->unused_chunks--;
-	}
-
-	return c;
-}
-
 static void chunkqueue_prepend_chunk(chunkqueue *cq, chunk *c) {
 	c->next = cq->first;
 	cq->first = c;
@@ -202,25 +227,25 @@ static void chunkqueue_append_chunk(chunkqueue *cq, chunk *c) {
 }
 
 static chunk * chunkqueue_prepend_mem_chunk(chunkqueue *cq) {
-    chunk *c = chunkqueue_get_unused_chunk(cq);
+    chunk *c = chunk_acquire();
     chunkqueue_prepend_chunk(cq, c);
     return c;
 }
 
 static chunk * chunkqueue_append_mem_chunk(chunkqueue *cq) {
-    chunk *c = chunkqueue_get_unused_chunk(cq);
+    chunk *c = chunk_acquire();
     chunkqueue_append_chunk(cq, c);
     return c;
 }
 
 static chunk * chunkqueue_append_file_chunk(chunkqueue *cq, buffer *fn, off_t offset, off_t len) {
-    chunk *c = chunkqueue_get_unused_chunk(cq);
+    chunk *c = chunk_acquire();
     chunkqueue_append_chunk(cq, c);
     c->type = FILE_CHUNK;
     c->file.start = offset;
     c->file.length = len;
     cq->bytes_in += len;
-    buffer_copy_buffer(c->file.name, fn);
+    buffer_copy_buffer(c->mem, fn);
     return c;
 }
 
@@ -231,7 +256,7 @@ void chunkqueue_reset(chunkqueue *cq) {
 
 	while (NULL != cur) {
 		chunk *next = cur->next;
-		chunkqueue_push_unused_chunk(cq, cur);
+		chunk_release(cur);
 		cur = next;
 	}
 
@@ -438,7 +463,7 @@ void chunkqueue_steal(chunkqueue *dest, chunkqueue *src, off_t len) {
 			/* drop empty chunk */
 			src->first = c->next;
 			if (c == src->last) src->last = NULL;
-			chunkqueue_push_unused_chunk(src, c);
+			chunk_release(c);
 			continue;
 		}
 
@@ -461,7 +486,7 @@ void chunkqueue_steal(chunkqueue *dest, chunkqueue *src, off_t len) {
 				break;
 			case FILE_CHUNK:
 				/* tempfile flag is in "last" chunk after the split */
-				chunkqueue_append_file(dest, c->file.name, c->file.start + c->offset, use);
+				chunkqueue_append_file(dest, c->mem, c->file.start + c->offset, use);
 				break;
 			}
 
@@ -564,7 +589,7 @@ int chunkqueue_append_mem_to_tempfile(server *srv, chunkqueue *dest, const char
 				dst_c->file.fd = -1;
 				if (0 != rc) {
 					log_error_write(srv, __FILE__, __LINE__, "sbss",
-						"close() temp-file", dst_c->file.name, "failed:",
+						"close() temp-file", dst_c->mem, "failed:",
 						strerror(errno));
 					return -1;
 				}
@@ -604,7 +629,7 @@ int chunkqueue_append_mem_to_tempfile(server *srv, chunkqueue *dest, const char
 			int retry = (errno == ENOSPC && dest->tempdirs && ++dest->tempdir_idx < dest->tempdirs->used);
 			if (!retry) {
 				log_error_write(srv, __FILE__, __LINE__, "sbs",
-						"write() temp-file", dst_c->file.name, "failed:",
+						"write() temp-file", dst_c->mem, "failed:",
 						strerror(errno));
 			}
 
@@ -616,7 +641,7 @@ int chunkqueue_append_mem_to_tempfile(server *srv, chunkqueue *dest, const char
 				dst_c->file.fd = -1;
 				if (0 != rc) {
 					log_error_write(srv, __FILE__, __LINE__, "sbss",
-						"close() temp-file", dst_c->file.name, "failed:",
+						"close() temp-file", dst_c->mem, "failed:",
 						strerror(errno));
 					return -1;
 				}
@@ -643,7 +668,7 @@ int chunkqueue_steal_with_tempfiles(server *srv, chunkqueue *dest, chunkqueue *s
 			/* drop empty chunk */
 			src->first = c->next;
 			if (c == src->last) src->last = NULL;
-			chunkqueue_push_unused_chunk(src, c);
+			chunk_release(c);
 			continue;
 		}
 
@@ -661,7 +686,7 @@ int chunkqueue_steal_with_tempfiles(server *srv, chunkqueue *dest, chunkqueue *s
 			} else {
 				/* partial chunk with length "use" */
 				/* tempfile flag is in "last" chunk after the split */
-				chunkqueue_append_file(dest, c->file.name, c->file.start + c->offset, use);
+				chunkqueue_append_file(dest, c->mem, c->file.start + c->offset, use);
 
 				c->offset += use;
 				force_assert(0 == len);
@@ -678,7 +703,7 @@ int chunkqueue_steal_with_tempfiles(server *srv, chunkqueue *dest, chunkqueue *s
 				/* finished chunk */
 				src->first = c->next;
 				if (c == src->last) src->last = NULL;
-				chunkqueue_push_unused_chunk(src, c);
+				chunk_release(c);
 			} else {
 				/* partial chunk */
 				c->offset += use;
@@ -720,8 +745,7 @@ void chunkqueue_mark_written(chunkqueue *cq, off_t len) {
 
 			cq->first = c->next;
 			if (c == cq->last) cq->last = NULL;
-
-			chunkqueue_push_unused_chunk(cq, c);
+			chunk_release(c);
 		} else { /* partial chunk */
 			c->offset += written;
 			written = 0;
@@ -741,8 +765,7 @@ void chunkqueue_remove_finished_chunks(chunkqueue *cq) {
 
 		cq->first = c->next;
 		if (c == cq->last) cq->last = NULL;
-
-		chunkqueue_push_unused_chunk(cq, c);
+		chunk_release(c);
 	}
 }
 
@@ -756,8 +779,7 @@ static void chunkqueue_remove_empty_chunks(chunkqueue *cq) {
 			chunk *empty = c->next;
 			c->next = empty->next;
 			if (empty == cq->last) cq->last = c;
-
-			chunkqueue_push_unused_chunk(cq, empty);
+			chunk_release(empty);
 		}
 	}
 }
@@ -775,8 +797,8 @@ int chunkqueue_open_file_chunk(server *srv, chunkqueue *cq) {
 	toSend = c->file.length - c->offset;
 
 	if (-1 == c->file.fd) {
-		if (-1 == (c->file.fd = fdevent_open_cloexec(c->file.name->ptr, O_RDONLY, 0))) {
-			log_error_write(srv, __FILE__, __LINE__, "ssb", "open failed:", strerror(errno), c->file.name);
+		if (-1 == (c->file.fd = fdevent_open_cloexec(c->mem->ptr, O_RDONLY, 0))) {
+			log_error_write(srv, __FILE__, __LINE__, "ssb", "open failed:", strerror(errno), c->mem);
 			return -1;
 		}
 	}
@@ -790,7 +812,7 @@ int chunkqueue_open_file_chunk(server *srv, chunkqueue *cq) {
 	}
 
 	if (offset > st.st_size || toSend > st.st_size || offset > st.st_size - toSend) {
-		log_error_write(srv, __FILE__, __LINE__, "sb", "file shrunk:", c->file.name);
+		log_error_write(srv, __FILE__, __LINE__, "sb", "file shrunk:", c->mem);
 		return -1;
 	}
 

src/chunk.h

@@ -43,9 +43,6 @@ typedef struct {
 	chunk *first;
 	chunk *last;
 
-	chunk *unused;
-	size_t unused_chunks;
-
 	off_t bytes_in, bytes_out;
 
 	array *tempdirs;
@@ -53,6 +50,12 @@ typedef struct {
 	unsigned int tempdir_idx;
 } chunkqueue;
 
+buffer * chunk_buffer_acquire(void);
+void chunk_buffer_release(buffer *b);
+
+void chunkqueue_chunk_pool_clear(void);
+void chunkqueue_chunk_pool_free(void);
+
 chunkqueue *chunkqueue_init(void);
 void chunkqueue_set_tempdirs_default_reset (void);
 void chunkqueue_set_tempdirs_default (array *tempdirs, unsigned int upload_temp_file_size);

src/server.c

@@ -398,6 +398,7 @@ static void server_free(server *srv) {
 	array_free(srv->split_vals);
 
 	li_rand_cleanup();
+	chunkqueue_chunk_pool_free();
 
 	free(srv);
 }
@@ -1792,6 +1793,8 @@ static int server_main (server * const srv, int argc, char **argv) {
 				}
 			      #endif
 
+				/* free excess chunkqueue buffers every 64 seconds */
+				if (0 == (min_ts & 0x3f)) chunkqueue_chunk_pool_clear();
 				/* cleanup stat-cache */
 				stat_cache_trigger_cleanup(srv);
 				/* reset global/aggregate rate limit counters */