[SERVER] Some sanity here and there, minor fixes, trying to track down proxy corruption

7 files changed, 54 insertions, 13 deletions

diff --git a/src/server/altservers.c b/src/server/altservers.c
index 641e36b..89c724f 100644
--- a/src/server/altservers.c
+++ b/src/server/altservers.c
@@ -377,8 +377,9 @@ static void *altserver_main(void *data)
 				if ( writev( sock, iov, 2 ) != len + sizeof(request) ) goto server_failed;
 				// See if selecting the image succeeded ++++++++++++++++++++++++++++++
 				if ( recv( sock, &reply, sizeof(reply), MSG_WAITALL ) != sizeof(reply) ) {
-					ERROR_GOTO_VA( server_failed, "[ERROR] Received corrupted reply header after CMD_SELECT_IMAGE (%s)",
-					        uplink->image->lower_name );
+					//ERROR_GOTO_VA( server_failed, "[ERROR] Received corrupted reply header after CMD_SELECT_IMAGE (%s)",
+					//        uplink->image->lower_name );
+					goto server_failed;
 				}
 				// check reply header
 				fixup_reply( reply );
@@ -419,7 +420,7 @@ static void *altserver_main(void *data)
 				fixup_reply( reply );
 				if ( reply.cmd != CMD_GET_BLOCK || reply.size != DNBD3_BLOCK_SIZE ) ERROR_GOTO_VA( server_failed,
 				        "[ERROR] Reply to random block request is %d bytes for %s", reply.size, uplink->image->lower_name );
-				if ( recv( sock, buffer, DNBD3_BLOCK_SIZE, 0 ) != DNBD3_BLOCK_SIZE ) ERROR_GOTO_VA( server_failed,
+				if ( recv( sock, buffer, DNBD3_BLOCK_SIZE, MSG_WAITALL ) != DNBD3_BLOCK_SIZE ) ERROR_GOTO_VA( server_failed,
 				        "[ERROR] Could not read random block from socket for %s", uplink->image->lower_name );
 				clock_gettime( CLOCK_MONOTONIC_RAW, &end );
 				// Measurement done - everything fine so far
@@ -450,7 +451,6 @@ static void *altserver_main(void *data)
 				uplink->rttTestResult = RTT_DOCHANGE;
 			} else {
 				// nope
-				printf( "DONT CHANGE: best: %uµs, current: %uµs\n", bestRtt, currentRtt );
 				if ( bestSock != -1 ) close( bestSock );
 				uplink->rttTestResult = RTT_DONTCHANGE;
 			}
diff --git a/src/server/image.c b/src/server/image.c
index 679dd40..af752a9 100644
--- a/src/server/image.c
+++ b/src/server/image.c
@@ -74,7 +74,7 @@ void image_update_cachemap(dnbd3_image_t *image, uint64_t start, uint64_t end, c
 	// This should always be block borders due to how the protocol works, but better be safe
 	// than accidentally mark blocks as cached when they really aren't entirely cached.
 	end &= ~(uint64_t)(DNBD3_BLOCK_SIZE - 1);
-	start = (start + DNBD3_BLOCK_SIZE - 1) & ~(uint64_t)(DNBD3_BLOCK_SIZE - 1);
+	start = (uint64_t)(start + DNBD3_BLOCK_SIZE - 1) & ~(uint64_t)(DNBD3_BLOCK_SIZE - 1);
 	int dirty = FALSE;
 	int pos = start;
 	spin_lock( &image->lock );
@@ -97,7 +97,7 @@ void image_update_cachemap(dnbd3_image_t *image, uint64_t start, uint64_t end, c
 		pos += DNBD3_BLOCK_SIZE;
 	}
 	spin_unlock( &image->lock );
-	if ( dirty ) {
+	if ( set && dirty ) {
 		// If dirty is set, at least one of the blocks was not cached before, so queue all hash blocks
 		// for checking, even though this might lead to checking some hash block again, if it was
 		// already complete and the block range spanned at least two hash blocks.
@@ -132,6 +132,10 @@ int image_save_cache_map(dnbd3_image_t *image)
 	if ( fd < 0 ) return FALSE;
 
 	write( fd, image->cache_map, ((image->filesize + (1 << 15) - 1) >> 15) * sizeof(char) );
+	if ( image->cacheFd != -1 ) {
+		fsync( image->cacheFd );
+	}
+	fsync( fd );
 	close( fd );
 
 	return TRUE;
@@ -240,8 +244,29 @@ void image_remove(dnbd3_image_t *image)
 }
 
 /**
+ * Kill all uplinks
+ */
+void image_killUplinks()
+{
+	int i;
+	spin_lock( &_images_lock );
+	for (i = 0; i < _num_images; ++i) {
+		if ( _images[i] == NULL ) continue;
+		spin_lock( &_images[i]->lock );
+		if ( _images[i]->uplink != NULL ) {
+			_images[i]->uplink->shutdown = TRUE;
+			if ( _images[i]->uplink->signal != -1 ) {
+				write( _images[i]->uplink->signal, "", 1 );
+			}
+		}
+		spin_unlock( &_images[i]->lock );
+	}
+	spin_unlock( &_images_lock );
+}
+
+/**
  * Free image. DOES NOT check if it's in use.
- * DOES NOT lock on anything.
+ * Indirectly locks on uplink.queueLock
  * DO NOT lock on the image when calling.
  */
 dnbd3_image_t* image_free(dnbd3_image_t *image)
@@ -249,11 +274,11 @@ dnbd3_image_t* image_free(dnbd3_image_t *image)
 	assert( image != NULL );
 	//
 	image_save_cache_map( image );
+	uplink_shutdown( image );
 	free( image->cache_map );
 	free( image->crc32 );
 	free( image->path );
 	free( image->lower_name );
-	uplink_shutdown( image );
 	if ( image->cacheFd != -1 ) close( image->cacheFd );
 	spin_destroy( &image->lock );
 	//
diff --git a/src/server/image.h b/src/server/image.h
index 9e70e6f..66d2563 100644
--- a/src/server/image.h
+++ b/src/server/image.h
@@ -18,6 +18,8 @@ dnbd3_image_t* image_get(char *name, uint16_t revision);
 
 void image_release(dnbd3_image_t *image);
 
+void image_killUplinks();
+
 dnbd3_image_t* image_free(dnbd3_image_t *image);
 
 int image_load_all(char *path);
diff --git a/src/server/net.c b/src/server/net.c
index b9fee3d..f0f9bf8 100644
--- a/src/server/net.c
+++ b/src/server/net.c
@@ -46,7 +46,7 @@ static inline char recv_request_header(int sock, dnbd3_request_t *request)
 	int ret;
 	// Read request header from socket
 	if ( (ret = recv( sock, request, sizeof(*request), MSG_WAITALL )) != sizeof(*request) ) {
-		if ( ret == 0 ) return 0;
+		if ( ret == 0 ) return FALSE;
 		printf( "[DEBUG] Error receiving request: Could not read message header (%d/%d)\n", ret, (int)sizeof(*request) );
 		return FALSE;
 	}
@@ -353,5 +353,6 @@ void *net_client_handler(void *dnbd3_client)
 	if ( image_file != -1 ) close( image_file );
 	dnbd3_remove_client( client );
 	client = dnbd3_free_client( client );
+	pthread_detach( client->thread );
 	return NULL ;
 }
diff --git a/src/server/server.c b/src/server/server.c
index 4bfa2d5..48a4f1e 100644
--- a/src/server/server.c
+++ b/src/server/server.c
@@ -114,6 +114,9 @@ void dnbd3_cleanup()
 	}
 	socket_count = 0;
 
+	// Terminate all uplinks
+	image_killUplinks();
+
 	// Clean up clients
 	spin_lock( &_clients_lock );
 	for (i = 0; i < _num_clients; ++i) {
@@ -317,7 +320,6 @@ int main(int argc, char *argv[])
 			dnbd3_client = dnbd3_free_client( dnbd3_client );
 			continue;
 		}
-		pthread_detach( dnbd3_client->thread );
 	}
 
 	dnbd3_cleanup();
diff --git a/src/server/sockhelper.c b/src/server/sockhelper.c
index dfbe5e3..e202d78 100644
--- a/src/server/sockhelper.c
+++ b/src/server/sockhelper.c
@@ -79,7 +79,7 @@ void sock_set_timeout(const int sockfd, const int milliseconds)
 {
 	struct timeval tv;
 	tv.tv_sec = milliseconds / 1000;
-	tv.tv_usec = milliseconds * 1000;
+	tv.tv_usec = (milliseconds * 1000) % 1000000;
 	setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
 	setsockopt(sockfd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
 }
diff --git a/src/server/uplink.c b/src/server/uplink.c
index 5eda88f..fea48a0 100644
--- a/src/server/uplink.c
+++ b/src/server/uplink.c
@@ -3,6 +3,7 @@
 #include "memlog.h"
 #include "sockhelper.h"
 #include "image.h"
+#include "helper.h"
 #include "altservers.h"
 #include <pthread.h>
 #include <sys/socket.h>
@@ -67,13 +68,13 @@ void uplink_shutdown(dnbd3_image_t *image)
 	spin_lock( &uplink->queueLock );
 	image->uplink = NULL;
 	uplink->shutdown = TRUE;
+	spin_unlock( &uplink->queueLock );
+	spin_destroy( &uplink->queueLock );
 	if ( uplink->signal != -1 ) write( uplink->signal, "", 1 );
 	if ( uplink->image != NULL ) {
 		pthread_join( uplink->thread, NULL );
 	}
 	free( uplink->recvBuffer );
-	spin_unlock( &uplink->queueLock );
-	spin_destroy( &uplink->queueLock );
 	free( uplink );
 }
 
@@ -238,6 +239,9 @@ static void* uplink_mainloop(void *data)
 			link->betterFd = -1;
 			link->currentServer = link->betterServer;
 			link->image->working = TRUE;
+			if ( host_to_string( &link->currentServer, buffer, sizeof buffer ) ) {
+				printf( "[DEBUG] Now connected to %s\n", buffer );
+			}
 			memset( &ev, 0, sizeof(ev) );
 			ev.events = EPOLLIN;
 			ev.data.fd = link->fd;
@@ -344,6 +348,10 @@ static void uplink_handle_receive(dnbd3_connection_t *link)
 		goto error_cleanup;
 	}
 	fixup_reply( reply );
+	if ( reply.magic != dnbd3_packet_magic ) {
+		memlogf( "[WARNING] Uplink server's packet did not start with dnbd3_packet_magic (%s)", link->image->path );
+		goto error_cleanup;
+	}
 	if ( reply.size > 9000000 ) {
 		memlogf( "[WARNING] Pure evil: Uplink server sent too much payload for %s", link->image->path );
 		goto error_cleanup;
@@ -363,6 +371,9 @@ static void uplink_handle_receive(dnbd3_connection_t *link)
 		done += ret;
 	}
 	// Payload read completely
+	// Bail out if we're not interested
+	if ( reply.cmd != CMD_GET_BLOCK ) return;
+	// Is a legit block reply
 	const uint64_t start = reply.handle;
 	const uint64_t end = reply.handle + reply.size;
 	// 1) Write to cache file