[SERVER] integrity: Group check requests, use sync_file_range()

This requires a much shorter queue and balances hashing between
different images if the checker lags behind.
On Linux, use sync_file_range() instead of fsync() before reading
back to speed up flushing.

1 files changed, 51 insertions, 25 deletions

diff --git a/src/server/integrity.c b/src/server/integrity.c
index 4d01ba7..88b7487 100644
--- a/src/server/integrity.c
+++ b/src/server/integrity.c
@@ -13,13 +13,15 @@
 #include <unistd.h>
 #include <fcntl.h>
 
-#define CHECK_QUEUE_SIZE 500
+#define CHECK_QUEUE_SIZE 200
+
+#define CHECK_ALL (0x7fffffff)
 
 typedef struct
 {
 	dnbd3_image_t *image; // Image to check
 	int block;            // Block to check
-	bool full;            // Check all blocks in image; .block will be increased
+	int count;            // How many blocks to check starting at .block
 } queue_entry;
 
 static pthread_t thread;
@@ -73,13 +75,23 @@ void integrity_shutdown()
  */
 void integrity_check(dnbd3_image_t *image, int block)
 {
+	if ( !bRunning ) {
+		logadd( LOG_MINOR, "Ignoring check request; thread not running..." );
+		return;
+	}
 	int i, freeSlot = -1;
 	pthread_mutex_lock( &integrityQueueLock );
 	for (i = 0; i < queueLen; ++i) {
 		if ( freeSlot == -1 && checkQueue[i].image == NULL ) {
 			freeSlot = i;
 		} else if ( checkQueue[i].image == image
-				&& ( checkQueue[i].block == block || checkQueue[i].full ) ) {
+				&& checkQueue[i].block <= block && checkQueue[i].block + checkQueue[i].count >= block ) {
+			// Already queued check dominates this one, or at least lies directly before this block
+			if ( checkQueue[i].block + checkQueue[i].count == block ) {
+				// It's directly before this one; expand range
+				checkQueue[i].count += 1;
+			}
+			logadd( LOG_DEBUG2, "Attaching to existing check request (%d/%d) (%d +%d)", i, queueLen, checkQueue[i].block, checkQueue[i].count );
 			pthread_mutex_unlock( &integrityQueueLock );
 			return;
 		}
@@ -87,7 +99,7 @@ void integrity_check(dnbd3_image_t *image, int block)
 	if ( freeSlot == -1 ) {
 		if ( queueLen >= CHECK_QUEUE_SIZE ) {
 			pthread_mutex_unlock( &integrityQueueLock );
-			logadd( LOG_DEBUG1, "Check queue full, discarding check request...\n" );
+			logadd( LOG_INFO, "Check queue full, discarding check request...\n" );
 			return;
 		}
 		freeSlot = queueLen++;
@@ -95,10 +107,10 @@ void integrity_check(dnbd3_image_t *image, int block)
 	checkQueue[freeSlot].image = image;
 	if ( block == -1 ) {
 		checkQueue[freeSlot].block = 0;
-		checkQueue[freeSlot].full = true;
+		checkQueue[freeSlot].count = CHECK_ALL;
 	} else {
 		checkQueue[freeSlot].block = block;
-		checkQueue[freeSlot].full = false;
+		checkQueue[freeSlot].count = 1;
 	}
 	pthread_cond_signal( &queueSignal );
 	pthread_mutex_unlock( &integrityQueueLock );
@@ -126,13 +138,13 @@ static void* integrity_main(void * data UNUSED)
 		for (i = queueLen - 1; i >= 0; --i) {
 			if ( _shutdown ) break;
 			dnbd3_image_t * const image = image_lock( checkQueue[i].image );
-			if ( !checkQueue[i].full || image == NULL ) {
-				checkQueue[i].image = NULL;
+			if ( checkQueue[i].count == 0 || image == NULL ) {
+				checkQueue[i].image = image_release( image );
 				if ( i + 1 == queueLen ) queueLen--;
+				continue;
 			}
-			if ( image == NULL ) continue;
 			// We have the image. Call image_release() some time
-			bool full = checkQueue[i].full;
+			const int qCount = checkQueue[i].count;
 			bool foundCorrupted = false;
 			spin_lock( &image->lock );
 			if ( image->crc32 != NULL && image->realFilesize != 0 ) {
@@ -158,19 +170,23 @@ static void* integrity_main(void * data UNUSED)
 					image_ensureOpen( image );
 					fd = image->readFd;
 				}
-				int checkCount = full ? 5 : 1;
+				int checkCount = MIN( qCount, 5 );
 				if ( fd != -1 ) {
 					while ( blocks[0] < numHashBlocks && !_shutdown ) {
 						const uint64_t start = blocks[0] * HASH_BLOCK_SIZE;
 						const uint64_t end = MIN( (uint64_t)(blocks[0] + 1) * HASH_BLOCK_SIZE, image->virtualFilesize );
 						bool complete = true;
-						if ( full ) {
+						if ( qCount == CHECK_ALL ) {
 							// When checking full image, skip incomplete blocks, otherwise assume block is complete
 							spin_lock( &image->lock );
 							complete = image_isHashBlockComplete( image->cache_map, blocks[0], fileSize );
 							spin_unlock( &image->lock );
 						}
+#if defined(linux) || defined(__linux)
+						if ( sync_file_range( fd, start, end - start, SYNC_FILE_RANGE_WAIT_BEFORE | SYNC_FILE_RANGE_WRITE | SYNC_FILE_RANGE_WAIT_AFTER ) == -1 ) {
+#else
 						if ( fsync( fd ) == -1 ) {
+#endif
 							logadd( LOG_ERROR, "Cannot flush %s for integrity check", image->path );
 							exit( 1 );
 						}
@@ -191,36 +207,46 @@ static void* integrity_main(void * data UNUSED)
 							logadd( LOG_WARNING, "Hash check for block %d of %s failed!", blocks[0], image->name );
 							image_updateCachemap( image, start, end, false );
 							// If this is not a full check, queue one
-							if ( !full ) {
+							if ( qCount != CHECK_ALL ) {
 								logadd( LOG_INFO, "Queueing full check for %s", image->name );
 								integrity_check( image, -1 );
 							}
 							foundCorrupted = true;
 						}
+						blocks[0]++; // Increase before break, so it always points to the next block to check after loop
 						if ( complete && --checkCount == 0 ) break;
-						blocks[0]++;
 					}
 					if ( direct ) {
 						close( fd );
 					}
 				}
 				pthread_mutex_lock( &integrityQueueLock );
-				if ( full ) {
-					assert( checkQueue[i].image == image );
-					assert( checkQueue[i].full );
-					if ( checkCount == 0 ) {
-						// Not done yet, keep going
-						checkQueue[i].block = blocks[0] + 1;
-					} else {
-						// Didn't check as many blocks as requested, so we must be done
-						checkQueue[i].image = NULL;
-						if ( i + 1 == queueLen ) queueLen--;
+				assert( checkQueue[i].image == image );
+				if ( qCount != CHECK_ALL ) {
+					// Not a full check; update the counter
+					checkQueue[i].count -= ( blocks[0] - checkQueue[i].block );
+					if ( checkQueue[i].count < 0 ) {
+						logadd( LOG_WARNING, "BUG! checkQueue counter ran negative" );
+					}
+				}
+				if ( checkCount > 0 || checkQueue[i].count <= 0 || fd == -1 ) {
+					// Done with this task as nothing left, OR we don't have an fd to read from
+					if ( fd == -1 ) {
+						logadd( LOG_WARNING, "Cannot hash check %s: bad fd", image->path );
+					}
+					checkQueue[i].image = NULL;
+					if ( i + 1 == queueLen ) queueLen--;
+					// Mark as working again if applicable
+					if ( !foundCorrupted ) {
 						spin_lock( &image->lock );
 						if ( image->uplink != NULL ) { // TODO: image_determineWorkingState() helper?
 							image->working = image->uplink->fd != -1 && image->readFd != -1;
 						}
 						spin_unlock( &image->lock );
 					}
+				} else {
+					// Still more blocks to go...
+					checkQueue[i].block = blocks[0];
 				}
 			} else {
 				spin_unlock( &image->lock );
@@ -243,6 +269,6 @@ static void* integrity_main(void * data UNUSED)
 	pthread_mutex_unlock( &integrityQueueLock );
 	if ( buffer != NULL ) free( buffer );
 	bRunning = false;
-	return NULL ;
+	return NULL;
 }