[*] Support hop-counting in request header, protocol version 3

We steal 8 bits from the request offset to count hops when requests
get relayed by proxies. This still leaves plenty of bits for the
offset (56 bits, supporting images of up to 72 petabytes).
This is used to detect proxy cycles. The algorithm is not perfect
but should prevent endless relays of the same request.
This is backwards compatible to old clients and servers, as the server
only ever sets the hopcount in relayed requests if the upstream server
is using protocol version 3 or newer, and clients are automatically
upwards compatible as there is practically no image larger than 74PB,
so the newly introduced hop count field is always 0 even in requests
from old clients.

1 files changed, 34 insertions, 13 deletions

diff --git a/src/server/uplink.c b/src/server/uplink.c
index a8d2f0b..65cefb5 100644
--- a/src/server/uplink.c
+++ b/src/server/uplink.c
@@ -43,7 +43,7 @@ uint64_t uplink_getTotalBytesReceived()
  * image. Uplinks run in their own thread.
  * Locks on: _images[].lock
  */
-bool uplink_init(dnbd3_image_t *image, int sock, dnbd3_host_t *host)
+bool uplink_init(dnbd3_image_t *image, int sock, dnbd3_host_t *host, int version)
 {
 	if ( !_isProxy ) return false;
 	dnbd3_connection_t *link = NULL;
@@ -72,6 +72,7 @@ bool uplink_init(dnbd3_image_t *image, int sock, dnbd3_host_t *host)
 		link->betterFd = sock;
 		link->betterServer = *host;
 		link->rttTestResult = RTT_DOCHANGE;
+		link->betterVersion = version;
 	} else {
 		link->betterFd = -1;
 		link->rttTestResult = RTT_IDLE;
@@ -145,7 +146,7 @@ void uplink_removeClient(dnbd3_connection_t *uplink, dnbd3_client_t *client)
  * Request a chunk of data through an uplink server
  * Locks on: image.lock, uplink.queueLock
  */
-bool uplink_request(dnbd3_client_t *client, uint64_t handle, uint64_t start, uint32_t length)
+bool uplink_request(dnbd3_client_t *client, uint64_t handle, uint64_t start, uint32_t length, uint8_t hops)
 {
 	if ( client == NULL || client->image == NULL ) return false;
 	spin_lock( &client->image->lock );
@@ -161,9 +162,10 @@ bool uplink_request(dnbd3_client_t *client, uint64_t handle, uint64_t start, uin
 		return false;
 	}
 	// Check if the client is the same host as the uplink. If so assume this is a circular proxy chain
-	if ( isSameAddress( &uplink->currentServer, &client->host ) ) {
+	// This might be a false positive if there are multiple instances running on the same host (IP)
+	if ( hops != 0 && isSameAddress( &uplink->currentServer, &client->host ) ) {
 		spin_unlock( &client->image->lock );
-		logadd( LOG_DEBUG1, "Proxy cycle detected" );
+		logadd( LOG_WARNING, "Proxy cycle detected (same host)." );
 		return false;
 	}
 
@@ -171,19 +173,34 @@ bool uplink_request(dnbd3_client_t *client, uint64_t handle, uint64_t start, uin
 	int existingType = -1; // ULR_* type of existing request
 	int i;
 	int freeSlot = -1;
+	bool requestLoop = false;
 	const uint64_t end = start + length;
 
 	spin_lock( &uplink->queueLock );
 	spin_unlock( &client->image->lock );
 	for (i = 0; i < uplink->queueLen; ++i) {
-		if ( freeSlot == -1 && uplink->queue[i].status == ULR_FREE ) freeSlot = i;
+		if ( freeSlot == -1 && uplink->queue[i].status == ULR_FREE ) {
+			freeSlot = i;
+			continue;
+		}
 		if ( uplink->queue[i].status != ULR_PENDING && uplink->queue[i].status != ULR_NEW ) continue;
-		if ( (foundExisting == -1 || existingType == ULR_PENDING) && uplink->queue[i].from <= start && uplink->queue[i].to >= end ) {
-			foundExisting = i;
-			existingType = uplink->queue[i].status;
-			break;
+		if ( uplink->queue[i].from <= start && uplink->queue[i].to >= end ) {
+			if ( hops > uplink->queue[i].hopCount ) {
+				requestLoop = true;
+				break;
+			}
+			if ( foundExisting == -1 || existingType == ULR_PENDING ) {
+				foundExisting = i;
+				existingType = uplink->queue[i].status;
+				if ( freeSlot != -1 ) break;
+			}
 		}
 	}
+	if ( requestLoop ) {
+		spin_unlock( &uplink->queueLock );
+		logadd( LOG_WARNING, "Rejecting relay of request to upstream proxy because of possible cyclic proxy chain. Incoming hop-count is %" PRIu8 ".", hops );
+		return false;
+	}
 	if ( freeSlot == -1 ) {
 		if ( uplink->queueLen >= SERVER_MAX_UPLINK_QUEUE ) {
 			spin_unlock( &uplink->queueLock );
@@ -198,7 +215,7 @@ bool uplink_request(dnbd3_client_t *client, uint64_t handle, uint64_t start, uin
 	// a race condition where the reply for the outstanding request already arrived and the uplink thread
 	// is currently traversing the request queue. As it is processing the queue from highest to lowest index, it might
 	// already have passed the index of the free slot we determined, but not reached the existing request we just found above.
-	if ( foundExisting != -1 && existingType != ULR_NEW && freeSlot > foundExisting ) foundExisting = -1;
+	if ( foundExisting != -1 && existingType != ULR_NEW && freeSlot > foundExisting ) foundExisting = -1; // -1 means "send request"
 #ifdef _DEBUG
 	if ( foundExisting != -1 ) {
 		logadd( LOG_DEBUG2, "%p (%s) Found existing request of type %s at slot %d, attaching in slot %d.\n", (void*)uplink, uplink->image->name, existingType == ULR_NEW ? "ULR_NEW" : "ULR_PENDING", foundExisting, freeSlot );
@@ -215,6 +232,7 @@ bool uplink_request(dnbd3_client_t *client, uint64_t handle, uint64_t start, uin
 	uplink->queue[freeSlot].client = client;
 	//int old = uplink->queue[freeSlot].status;
 	uplink->queue[freeSlot].status = (foundExisting == -1 ? ULR_NEW : ULR_PENDING);
+	uplink->queue[freeSlot].hopCount = hops;
 #ifdef _DEBUG
 	uplink->queue[freeSlot].entered = time( NULL );
 	//logadd( LOG_DEBUG2 %p] Inserting request at slot %d, was %d, now %d, handle %" PRIu64 ", Range: %" PRIu64 "-%" PRIu64 "\n", (void*)uplink, freeSlot, old, uplink->queue[freeSlot].status, uplink->queue[freeSlot, ".handle, start, end );
@@ -272,6 +290,7 @@ static void* uplink_mainloop(void *data)
 			link->fd = link->betterFd;
 			link->betterFd = -1;
 			link->currentServer = link->betterServer;
+			link->version = link->betterVersion;
 			spin_unlock( &link->rttLock );
 			discoverFailCount = 0;
 			if ( fd != -1 ) close( fd );
@@ -439,8 +458,10 @@ static void uplink_sendRequests(dnbd3_connection_t *link, bool newOnly)
 		link->queue[j].status = ULR_PENDING;
 		const uint64_t offset = link->queue[j].from;
 		const uint32_t size = link->queue[j].to - link->queue[j].from;
+		uint8_t hops = link->queue[j].hopCount;
 		spin_unlock( &link->queueLock );
-		const int ret = dnbd3_get_block( link->fd, offset, size, offset );
+		if ( hops < 200 ) hops += 1;
+		const int ret = dnbd3_get_block( link->fd, offset, size, offset, COND_HOPCOUNT( link->version, hops ) );
 		if ( !ret ) {
 			// Non-critical - if the connection dropped or the server was changed
 			// the thread will re-send this request as soon as the connection
@@ -488,7 +509,7 @@ static void uplink_sendReplicationRequest(dnbd3_connection_t *link)
 		// Unlocked - do not break or continue here...
 		const uint64_t offset = link->replicationHandle = (uint64_t)i * (uint64_t)requestBlockSize;
 		const uint32_t size = MIN( image->realFilesize - offset, requestBlockSize );
-		if ( !dnbd3_get_block( link->fd, offset, size, link->replicationHandle ) ) {
+		if ( !dnbd3_get_block( link->fd, offset, size, link->replicationHandle, COND_HOPCOUNT( link->version, 1 ) ) ) {
 			logadd( LOG_DEBUG1, "Error sending background replication request to uplink server!\n" );
 			return;
 		}
@@ -644,7 +665,7 @@ static void uplink_handleReceive(dnbd3_connection_t *link)
  */
 static int uplink_sendKeepalive(const int fd)
 {
-	static dnbd3_request_t request = { 0, 0, 0, 0, 0 };
+	static dnbd3_request_t request = { 0 };
 	if ( request.magic == 0 ) {
 		request.magic = dnbd3_packet_magic;
 		request.cmd = CMD_KEEPALIVE;