test reliability, shutdown race fixes and continued work on monitoring

Author: splitice

.github/workflows/build-test.yml

@@ -1,6 +1,9 @@
 name: build and test
 
-on: [push]
+on:
+  push:
+  schedule:
+    - cron:  '*/10 * * * *'
 
 
 jobs:
@@ -24,7 +27,7 @@ jobs:
       run: |
         PWD=$(pwd)
         sudo apt-get install valgrind
-        valgrind --leak-check=full ./tests/${{ matrix.release }}/tests $PWD/src/server/${{ matrix.release }}/scache $PWD/testcases/
+        valgrind --leak-check=full --show-leak-kinds=all ./tests/${{ matrix.release }}/tests $PWD/src/server/${{ matrix.release }}/scache $PWD/testcases/
     - name: Spam test
       run: |
         echo "todo"
@@ -49,4 +52,4 @@ jobs:
 
         query &
 
-        timeout 2m valgrind --leak-check=full --show-leak-kinds=all src/server/${{ matrix.release }}/scache -B 127.0.0.1:8082
+        timeout --preserve-status 2m valgrind --leak-check=full --show-leak-kinds=all src/server/${{ matrix.release }}/scache -B 127.0.0.1:8082

src/core/Makefile

@@ -34,7 +34,7 @@ error:
 	$(error Invalid configuration, please check your inputs)
 endif
 
-SOURCEFILES := connection.cpp db.cpp hash.cpp http.cpp http_parse.cpp http_parse_cache.cpp http_parse_mon.c http_respond.cpp read_buffer.cpp settings.cpp timer.cpp
+SOURCEFILES := connection.cpp db.cpp hash.cpp http.cpp http_parse.cpp http_parse_cache.cpp http_parse_mon.c http_respond.cpp read_buffer.cpp settings.cpp timer.cpp signal_handle.cpp
 EXTERNAL_LIBS := 
 EXTERNAL_LIBS_COPIED := $(foreach lib, $(EXTERNAL_LIBS),$(BINARYDIR)/$(notdir $(lib)))
 

src/core/connection.cpp

@@ -39,6 +39,9 @@ struct scache_connection_node ctable[CONNECTION_HASH_ENTRIES] = { 0 };
 int epfd;
 volatile sig_atomic_t stop_soon = 0;
 
+bool connection_stop_soon(){
+	return stop_soon != 0;
+}
 
 struct connection_thread_arg
 {
@@ -128,6 +131,8 @@ static int connection_non_blocking(int fd)
 
 void connection_close_listeners() {
 	int fd;
+	
+	WARN("Closing %d listeners", scache_listeners.listener_count);
 
 	for (uint32_t i = 0; i < scache_listeners.listener_count; i++)
 	{
@@ -242,23 +247,39 @@ int connection_open_listener(struct scache_bind ibind) {
 	return -1;
 }
 
-static scache_connection* connection_add(int fd) {
+static scache_connection* connection_add(int fd, listener_type client_type) {
 	scache_connection_node* node = &ctable[CONNECTION_HASH_KEY(fd)];
+	scache_connection_node* newNode = NULL;
+
 	if (node->connection.client_sock != -1) {
 		while (node->next != NULL) {
 			assert(node->connection.client_sock != -1);
 			node = node->next;
 		}
 
-		scache_connection_node* newNode = (scache_connection_node*)malloc(sizeof(scache_connection_node));
-		node->next = newNode;
-		node = newNode;
+		newNode = (scache_connection_node*)malloc(sizeof(scache_connection_node));
+	}else{
+		newNode = node;
 	}
 
 	// Initialize connection
-	memset(node, 0, sizeof(node)); /* .connection = {}, .next = NULL */
-	rbuf_init(&(node->connection.input));
-	node->connection.client_sock = fd;
+	memset(newNode, 0, sizeof(*newNode)); /* .connection = {}, .next = NULL */
+
+	// safe to set first
+	newNode->connection.ltype = client_type;
+	if(client_type == cache_listener){
+		newNode->connection.cache.target.key.fd = -1;
+	}
+	rbuf_init(&(newNode->connection.input));
+
+	// do last as marks connection slot as used
+	newNode->connection.client_sock = fd;
+
+	// this is a chained connection
+	if(node != newNode){
+		node->next = newNode;
+		node = newNode;
+	}
 
 	return &(node->connection);
 }
@@ -383,6 +404,7 @@ static void* connection_handle_accept(void *arg)
 				goto end;
 			}
 		} while(nfds == 0 && !stop_soon);
+		assert(nfds >= 0);
 		int n = 0;
 		int state = 1;
 		while (n < nfds) {
@@ -394,67 +416,63 @@ static void* connection_handle_accept(void *arg)
 			else if (events[n].events & EPOLLIN)
 			{
 				DEBUG("[#] Accepting connection from fd %d of type %s\n", fd, listener_type_string(our_type));
-				int client_sock;
-					
-				do
-				{
-					client_sock = accept(fd, NULL, NULL);
+				int client_sock = accept(fd, NULL, NULL);
 
-					if (client_sock < 0) {
-						if (errno == EAGAIN || errno == EWOULDBLOCK)
-						{
-							break;
-						}
+				if (client_sock < 0) {
+					if (errno != EAGAIN && errno != EWOULDBLOCK)
+					{
 						WARN("[#] accept() failed on fd %d of type %s. Error: %s", fd, listener_type_string(our_type), strerror(errno));
-						break;
-					}
-					else {
-						DEBUG("[#] Accepted connection %d on fd %d of type %s\n", client_sock, fd, listener_type_string(our_type));
-
-						//Connection will be non-blocking
-						if (connection_non_blocking(client_sock) < 0)
-							PFATAL("Setting connection to non blocking failed on fd %d of type %s.", fd, listener_type_string(our_type));
-						
-						//Enable TCP CORK
-						setsockopt(client_sock, IPPROTO_TCP, TCP_CORK, &state, sizeof(state));
-						
-						connections_queued* q = (connections_queued*)malloc(sizeof(connections_queued));
-						if(q == NULL){
-							close(client_sock);
-							WARN("[#] failed to allocate memory for connection. Abandoning incoming connection.");
-							continue;
-						}
-						q->client_sock = client_sock;
-						q->client_type = our_type;
-						q->next = NULL;
-						
-						// Insert connection into queue
-						pthread_mutex_lock(&cq_lock);
-						if (cq_tail == NULL)
-						{
-							assert(cq_head == NULL);
-							cq_tail = q;
-							cq_head = q;
-						}
-						else
-						{
-							cq_tail->next = q;
-							cq_tail = q;
-						}
-						pthread_mutex_unlock(&cq_lock);
-						
-						//Write a signal
-						int res;
-						do {
-							res = write(eventfd, &u, sizeof(u));
-							if(res == -1){
-								PFATAL("Unable to write to eventfd");
-							}
-							assert(res == sizeof(u));
-						} while(!res);
 					}
 					n++;
-				} while (!stop_soon);
+					continue;
+				}
+				else {
+					DEBUG("[#] Accepted connection %d on fd %d of type %s\n", client_sock, fd, listener_type_string(our_type));
+
+					//Connection will be non-blocking
+					if (connection_non_blocking(client_sock) < 0)
+						PFATAL("Setting connection to non blocking failed on fd %d of type %s.", fd, listener_type_string(our_type));
+					
+					//Enable TCP CORK
+					setsockopt(client_sock, IPPROTO_TCP, TCP_CORK, &state, sizeof(state));
+					
+					connections_queued* q = (connections_queued*)malloc(sizeof(connections_queued));
+					if(q == NULL){
+						close(client_sock);
+						n++;
+						WARN("[#] failed to allocate memory for connection. Abandoning incoming connection.");
+						continue;
+					}
+					q->client_sock = client_sock;
+					q->client_type = our_type;
+					q->next = NULL;
+					
+					// Insert connection into queue
+					pthread_mutex_lock(&cq_lock);
+					if (cq_tail == NULL)
+					{
+						assert(cq_head == NULL);
+						cq_tail = q;
+						cq_head = q;
+					}
+					else
+					{
+						cq_tail->next = q;
+						cq_tail = q;
+					}
+					pthread_mutex_unlock(&cq_lock);
+					
+					//Write a signal
+					int res;
+					do {
+						res = write(eventfd, &u, sizeof(u));
+						if(res == -1){
+							PFATAL("Unable to write to eventfd");
+						}
+						assert(res == sizeof(u));
+					} while(!res);
+				}
+				n++;
 			}
 		}
 	}
@@ -481,7 +499,7 @@ void connection_event_loop(void (*connection_handler)(scache_connection* connect
 	}
 
 	//Init Acceptor thread
-	connection_thread_arg* thread_arg = (connection_thread_arg*)malloc(sizeof(connection_thread_arg)  * 2) ;
+	connection_thread_arg thread_arg[2];
 	efd = eventfd(0, 0);
 	thread_arg[0].type = cache_listener;
 	thread_arg[1].type = mon_listener;
@@ -509,7 +527,6 @@ void connection_event_loop(void (*connection_handler)(scache_connection* connect
 					continue;
 				}
 				PFATAL("epoll_wait() failed");
-				goto end;
 			}
 		} while(nfds == 0 && !stop_soon);
 
@@ -521,7 +538,7 @@ void connection_event_loop(void (*connection_handler)(scache_connection* connect
 			{				
 				res = read(fd, &u, sizeof(u));
 
-				while (true)
+				while (!stop_soon)
 				{					
 					//Dequeue
 					pthread_mutex_lock(&cq_lock);
@@ -549,9 +566,8 @@ void connection_event_loop(void (*connection_handler)(scache_connection* connect
 
 					//Handle connection
 					DEBUG("[#%d] A new %s socket was accepted %d\n", fd, listener_type_string(client_type), client_sock);
-					scache_connection* connection = connection_add(client_sock);
+					scache_connection* connection = connection_add(client_sock, client_type);
 					assert(connection->client_sock == client_sock);
-					connection->ltype = client_type;
 					connection_handler(connection);
 
 					//Add socket to epoll
@@ -628,12 +644,16 @@ void connection_event_loop(void (*connection_handler)(scache_connection* connect
 end:
 	close(epfd);
 
-	pthread_join(tid[0], NULL);
-	pthread_join(tid[1], NULL);
+	errno = pthread_join(tid[0], NULL);
+	if(errno != 0) {
+		PFATAL("failed to join cache thread");
+	}
+	errno = pthread_join(tid[1], NULL);
+	if(errno != 0) {
+		PFATAL("failed to join mon thread");
+	}
 
 	close(efd);
-	
-	free(thread_arg);
 }
 
 /*

src/core/connection.h

@@ -12,6 +12,7 @@ void connection_event_loop(void(*connection_handler)(scache_connection* connecti
 void connection_setup(struct scache_binds cache_binds, struct scache_binds cache_monitor);
 void connection_cleanup();
 bool connection_remove(int fd);
+bool connection_stop_soon();
 
 #define _DEBUG_CONNECTION_HANDLER
 #ifdef DEBUG_CONNECTION_HANDLER

src/core/connection_structures.h

@@ -8,7 +8,7 @@ struct cache_target {
 	struct cache_entry* entry;
 	off64_t position;
 	off64_t end_position;
-	int fd;
+	int fd; // fd can be init to -1 because its not overlapping with table_target
 };
 
 struct table_target {

src/core/db.cpp

@@ -36,6 +36,7 @@ LRU
 #include "hash.h"
 #include "settings.h"
 #include "timer.h"
+#include "signal_handle.h"
 
 #ifdef DEBUG_BUILD
 #include <set>
@@ -734,17 +735,32 @@ bool db_open(const char* path) {
 
 	// Calculate the number of blocks in the blockfile
 	off64_t size = lseek64(db.fd_blockfile, 0L, SEEK_END);
+	if(size < 0){
+		PWARN("seek to end failed");
+		size = 0;
+		if(ftruncate(db.fd_blockfile, size) == -1){
+			PFATAL("Failed to completely truncate blockfile %s", db.path_blockfile);
+		}
+	}
 	db.blocks_exist = (uint32_t)(size / BLOCK_LENGTH);
+	lseek64(db.fd_blockfile, 0L, SEEK_SET);
 
 	// Mark all blocks that already exist in the block file as non-allocated
 	if(will_black){
 		if(size > (BLOCK_MAX_LOAD * BLOCK_LENGTH)) {
 			size = BLOCK_MAX_LOAD * BLOCK_LENGTH;
 			if(ftruncate(db.fd_blockfile, size) == -1){
-				PFATAL("Failed to truncate blockfile: %s", db.path_blockfile);
+				PFATAL("Failed to truncate blockfile %s", db.path_blockfile);
 			}
 			db.blocks_exist = (uint32_t)(size / BLOCK_LENGTH);
 		}
+		if((size % BLOCK_LENGTH) != 0){
+			WARN("Block file was strange size of %d", size);
+			
+			if(ftruncate(db.fd_blockfile, (size / BLOCK_LENGTH) * BLOCK_LENGTH) == -1){
+				PFATAL("Failed to truncate blockfile %s", db.path_blockfile);
+			}
+		}
 		for (off64_t i = 0; i < size; i += BLOCK_LENGTH) {
 			db_block_free((uint32_t)(i / BLOCK_LENGTH));
 		}
@@ -1363,6 +1379,7 @@ static pid_t db_index_flush(bool copyOnWrite){
 	if(copyOnWrite){
 		pid = fork();
 		if(pid != 0) return pid; // includes -1
+		signal_handler_remove();
 	}
 
 	//Open temporary index file

src/core/http_parse_mon.cpp

@@ -295,11 +295,11 @@ void monitoring_destroy(scache_connection* connection){
 	if(t != NULL){
 		assert(t->monitoring.prev == connection);
 		t->monitoring.prev = connection->monitoring.prev;
-		if(t->monitoring.prev == NULL) {
+		if(t->monitoring.prev == NULL && mon_head == connection) {
 			assert(mon_head == connection);
 			mon_head = t;
 		}
-	}else{
+	}else if(mon_tail == connection){
 		// Only if we are destroying the tail t will be NULL
 		assert(connection == mon_tail);
 		mon_tail = t;
@@ -312,13 +312,12 @@ void monitoring_destroy(scache_connection* connection){
 	if(t != NULL){
 		assert(t->monitoring.next == connection);
 		t->monitoring.next = connection->monitoring.next;
-		if(t->monitoring.next == NULL) {
+		if(t->monitoring.next == NULL && mon_tail == connection) {
 			assert(mon_tail == connection);
 			mon_tail = t;
 		}
-	}else{
+	}else if(mon_head == connection){
 		// Only if we are destroying the head will t be null
-		assert(mon_head == connection);
 		mon_head = t;
 	}
 }

src/core/http_respond.cpp

@@ -62,11 +62,10 @@ state_action http_respond_contentlength(scache_connection* connection) {
 }
 
 bool http_register_read(scache_connection* connection) {
-	int extern stop_soon;
 
 	bool res = connection_register_read(connection->client_sock);
 
-	if (!stop_soon && rbuf_write_remaining(&connection->input)) {
+	if (!connection_stop_soon() && rbuf_write_remaining(&connection->input)) {
 		http_read_handle(connection);
 	}