vine: properly replicate temp files (#4004)

* print rep count

* init

* update

* up

* up

* up

* lint

* up

* limit the max number of rep attempts

* update

Author: JinZhou5042

taskvine/src/bindings/python3/ndcctools/taskvine/manager.py

@@ -1766,6 +1766,14 @@ def log_txn_app(self, entry):
     def log_debug_app(self, entry):
         cvine.vine_log_debug_app(self._taskvine, entry)
 
+    ##
+    # Gets the number of replicas of a file.
+    #
+    # @param self   The manager to register this file
+    # @param file   The File object
+    def get_file_replica_count(self, file):
+        return cvine.vine_file_replica_count(self._taskvine, file._file)
+
 
 ##
 # @class ndcctools.taskvine.manager.Factory

taskvine/src/manager/taskvine.h

@@ -730,6 +730,13 @@ const char *vine_file_source(struct vine_file *f);
 */
 vine_file_type_t vine_file_type(struct vine_file *f);
 
+/** Get the number of replicas of a file.
+@param m A manager object
+@param f A file object.
+@return The number of replicas of the file.
+*/
+int vine_file_replica_count(struct vine_manager *m, struct vine_file *f);
+
 /** Declare a file object from a local file
 @param m A manager object
 @param source The path of the file on the local filesystem

taskvine/src/manager/vine_current_transfers.h

@@ -8,7 +8,7 @@ See the file COPYING for details.
 #include "uuid.h"
 
 #define VINE_FILE_SOURCE_MAX_TRANSFERS 1
-#define VINE_WORKER_SOURCE_MAX_TRANSFERS 3 // static 1 until if/when multiple transfer ports are opened up on worker transfer server
+#define VINE_WORKER_SOURCE_MAX_TRANSFERS 10
 
 char *vine_current_transfers_add(struct vine_manager *q, struct vine_worker_info *to, struct vine_worker_info *source_worker, const char *source_url);
 
@@ -24,7 +24,6 @@ int vine_current_transfers_url_in_use(struct vine_manager *q, const char *source
 
 int vine_current_transfers_dest_in_use(struct vine_manager *q,struct vine_worker_info *w);
 
-
 int vine_current_transfers_wipe_worker(struct vine_manager *q, struct vine_worker_info *w);
 
 void vine_current_transfers_print_table(struct vine_manager *q);

taskvine/src/manager/vine_file_replica_table.c

@@ -77,6 +77,12 @@ struct vine_file_replica *vine_file_replica_table_lookup(struct vine_worker_info
 	return hash_table_lookup(w->current_files, cachename);
 }
 
+// count the number of in-cluster replicas of a file
+int vine_file_replica_count(struct vine_manager *m, struct vine_file *f)
+{
+	return set_size(hash_table_lookup(m->file_worker_table, f->cached_name));
+}
+
 // find a worker (randomly) in posession of a specific file, and is ready to transfer it.
 struct vine_worker_info *vine_file_replica_table_find_worker(struct vine_manager *q, const char *cachename)
 {
@@ -124,40 +130,21 @@ struct vine_worker_info *vine_file_replica_table_find_worker(struct vine_manager
 }
 
 // trigger replications of file to satisfy temp_replica_count
-int vine_file_replica_table_replicate(struct vine_manager *m, struct vine_file *f)
+int vine_file_replica_table_replicate(struct vine_manager *m, struct vine_file *f, struct set *sources, int to_find)
 {
-	/* the number of replicated copies in this round */
-	int round_replication_count = 0;
-
-	if (vine_current_transfers_get_table_size(m) >= hash_table_size(m->worker_table) * m->worker_source_max_transfers) {
-		return round_replication_count;
-	}
-
-	struct set *sources = hash_table_lookup(m->file_worker_table, f->cached_name);
-	if (!sources) {
-		return round_replication_count;
-	}
-
 	int nsources = set_size(sources);
-	int to_find = MIN(m->temp_replica_count - nsources, m->transfer_replica_per_cycle);
-	if (to_find < 1) {
-		return round_replication_count;
-	}
-
-	debug(D_VINE, "Found %d workers holding %s, %d replicas needed", nsources, f->cached_name, to_find);
+	int round_replication_request_sent = 0;
 
 	/* get the elements of set so we can insert new replicas to sources */
 	struct vine_worker_info **sources_frozen = (struct vine_worker_info **)set_values(sources);
 	struct vine_worker_info *source;
 
-	int i = 0;
-	for (source = sources_frozen[i]; i < nsources; i++) {
-		if (round_replication_count >= to_find) {
-			break;
-		}
+	for (int i = 0; i < nsources; i++) {
 
-		int found_per_source = 0;
+		source = sources_frozen[i];
+		int dest_found = 0;
 
+		// skip if the file on the source is not ready to transfer
 		struct vine_file_replica *replica = hash_table_lookup(source->current_files, f->cached_name);
 		if (!replica || replica->state != VINE_FILE_REPLICA_STATE_READY) {
 			continue;
@@ -166,51 +153,60 @@ int vine_file_replica_table_replicate(struct vine_manager *m, struct vine_file *
 		char *source_addr = string_format("%s/%s", source->transfer_url, f->cached_name);
 		int source_in_use = vine_current_transfers_source_in_use(m, source);
 
+		// skip if the source is busy with other transfers
+		if (source_in_use >= m->worker_source_max_transfers) {
+			continue;
+		}
+
 		char *id;
-		struct vine_worker_info *peer;
+		struct vine_worker_info *dest;
 		int offset_bookkeep;
-		HASH_TABLE_ITERATE_RANDOM_START(m->worker_table, offset_bookkeep, id, peer)
-		{
 
-			if (found_per_source >= MIN(m->file_source_max_transfers, to_find)) {
-				break;
-			}
-
-			if (source_in_use >= m->worker_source_max_transfers) {
-				break;
-			}
-
-			if (!peer->transfer_port_active) {
+		HASH_TABLE_ITERATE_RANDOM_START(m->worker_table, offset_bookkeep, id, dest)
+		{
+			// skip if the source and destination are on the same host
+			if (set_lookup(sources, dest) || strcmp(source->hostname, dest->hostname) == 0) {
 				continue;
 			}
 
-			if (set_lookup(sources, peer)) {
+			// skip if the destination is not ready to transfer
+			if (!dest->transfer_port_active) {
 				continue;
 			}
 
-			if (vine_current_transfers_dest_in_use(m, peer) >= m->worker_source_max_transfers) {
+			// skip if the destination is busy with other transfers
+			if (vine_current_transfers_dest_in_use(m, dest) >= m->worker_source_max_transfers) {
 				continue;
 			}
 
-			if (strcmp(source->hostname, peer->hostname) == 0) {
-				continue;
-			}
+			debug(D_VINE, "replicating %s from %s to %s", f->cached_name, source->addrport, dest->addrport);
 
-			debug(D_VINE, "replicating %s from %s to %s", f->cached_name, source->addrport, peer->addrport);
+			vine_manager_put_url_now(m, dest, source_addr, f);
 
-			vine_manager_put_url_now(m, peer, source_addr, f);
+			round_replication_request_sent++;
 
-			source_in_use++;
-			found_per_source++;
-			round_replication_count++;
+			// break if we have found enough destinations for this source
+			if (++dest_found >= MIN(m->file_source_max_transfers, to_find)) {
+				break;
+			}
+
+			// break if the source becomes busy with transfers
+			if (++source_in_use >= m->worker_source_max_transfers) {
+				break;
+			}
 		}
 
 		free(source_addr);
+
+		// break if we have sent enough replication requests for this file
+		if (round_replication_request_sent >= to_find) {
+			break;
+		}
 	}
 
 	free(sources_frozen);
 
-	return round_replication_count;
+	return round_replication_request_sent;
 }
 
 /*

taskvine/src/manager/vine_file_replica_table.h

@@ -14,6 +14,7 @@ See the file COPYING for details.
 #define VINE_FILE_REPLICA_TABLE_H
 
 #include "taskvine.h"
+#include "set.h"
 #include "vine_file_replica.h"
 #include "vine_worker_info.h"
 
@@ -27,7 +28,7 @@ struct vine_file_replica *vine_file_replica_table_lookup(struct vine_worker_info
 
 struct vine_worker_info *vine_file_replica_table_find_worker(struct vine_manager *q, const char *cachename);
 
-int vine_file_replica_table_replicate(struct vine_manager *q, struct vine_file *f);
+int vine_file_replica_table_replicate(struct vine_manager *q, struct vine_file *f, struct set *sources, int to_find);
 
 int vine_file_replica_table_exists_somewhere( struct vine_manager *q, const char *cachename );
 

taskvine/src/manager/vine_manager.c

@@ -934,12 +934,15 @@ static void cleanup_worker(struct vine_manager *q, struct vine_worker_info *w)
 }
 
 /* Start replicating files that may need replication */
-
-static int recover_temp_files(struct vine_manager *q)
+static int consider_tempfile_replications(struct vine_manager *q)
 {
+	if (hash_table_size(q->temp_files_to_replicate) <= 0) {
+		return 0;
+	}
+
 	char *cached_name = NULL;
 	void *empty_val = NULL;
-	int total_replication_count = 0;
+	int total_replication_request_sent = 0;
 
 	static char key_start[PATH_MAX] = "random init";
 	int iter_control;
@@ -949,32 +952,55 @@ static int recover_temp_files(struct vine_manager *q)
 	{
 		struct vine_file *f = hash_table_lookup(q->file_table, cached_name);
 
-		if (f) {
-			int round_replication_count = vine_file_replica_table_replicate(q, f);
-
-			/* Worker busy or no replicas found */
-			if (round_replication_count < 1) {
-				/*
-				If no replicas are found, it indicates that the file doesn't exist, either pruned or lost.
-				Because a pruned file is removed from the recovery queue, so it definitely indicates that the file is lost.
-				*/
-				if (!vine_file_replica_table_exists_somewhere(q, f->cached_name) && q->transfer_temps_recovery) {
-					vine_manager_consider_recovery_task(q, f, f->recovery_task);
-				}
-				hash_table_remove(q->temp_files_to_replicate, cached_name);
-			} else {
-				if (iter_count_var > q->attempt_schedule_depth) {
-					strncpy(key_start, cached_name, PATH_MAX - 1);
-					key_start[PATH_MAX - 1] = '\0';
-					break;
-				}
+		if (!f) {
+			continue;
+		}
+
+		/* are there any available sources? */
+		struct set *sources = hash_table_lookup(q->file_worker_table, f->cached_name);
+		if (!sources) {
+			/* If no sources found, it indicates that the file doesn't exist, either pruned or lost.
+			Because a pruned file is removed from the recovery queue, so it definitely indicates that the file is lost. */
+			if (q->transfer_temps_recovery) {
+				vine_manager_consider_recovery_task(q, f, f->recovery_task);
+			}
+			hash_table_remove(q->temp_files_to_replicate, f->cached_name);
+			continue;
+		}
+
+		/* at least one source is able to transfer? */
+		int has_valid_source = 0;
+		struct vine_worker_info *s;
+		SET_ITERATE(sources, s)
+		{
+			if (s->transfer_port_active && vine_current_transfers_source_in_use(q, s) < q->worker_source_max_transfers) {
+				has_valid_source = 1;
+				break;
 			}
+		}
+		if (!has_valid_source) {
+			continue;
+		}
+
+		/* has this file been fully replicated? */
+		int nsources = set_size(sources);
+		int to_find = MIN(q->temp_replica_count - nsources, q->transfer_replica_per_cycle);
+		if (to_find <= 0) {
+			hash_table_remove(q->temp_files_to_replicate, f->cached_name);
+			continue;
+		}
+
+		debug(D_VINE, "Found %d workers holding %s, %d replicas needed", nsources, f->cached_name, to_find);
+
+		int round_replication_request_sent = vine_file_replica_table_replicate(q, f, sources, to_find);
+		total_replication_request_sent += round_replication_request_sent;
 
-			total_replication_count += round_replication_count;
+		if (total_replication_request_sent >= q->attempt_schedule_depth) {
+			break;
 		}
 	}
 
-	return total_replication_count;
+	return total_replication_request_sent;
 }
 
 /* Insert into hashtable temp files that may need replication. */
@@ -5118,6 +5144,16 @@ static struct vine_task *vine_wait_internal(struct vine_manager *q, int timeout,
 			}
 		}
 
+		// Check if any temp files need replication and start replicating
+		BEGIN_ACCUM_TIME(q, time_internal);
+		result = consider_tempfile_replications(q);
+		END_ACCUM_TIME(q, time_internal);
+		if (result) {
+			// recovered at least one temp file
+			events++;
+			continue;
+		}
+
 		// send keepalives to appropriate workers
 		BEGIN_ACCUM_TIME(q, time_status_msgs);
 		ask_for_workers_updates(q);
@@ -5142,16 +5178,6 @@ static struct vine_task *vine_wait_internal(struct vine_manager *q, int timeout,
 			continue;
 		}
 
-		// Check if any temp files need replication and start replicating
-		BEGIN_ACCUM_TIME(q, time_internal);
-		result = recover_temp_files(q);
-		END_ACCUM_TIME(q, time_internal);
-		if (result) {
-			// recovered at least one temp file
-			events++;
-			continue;
-		}
-
 		if (q->process_pending_check) {
 			BEGIN_ACCUM_TIME(q, time_internal);
 			int pending = process_pending();
@@ -6106,9 +6132,7 @@ void vine_prune_file(struct vine_manager *m, struct vine_file *f)
 		}
 	}
 
-	/*
-	Pruned files do not need to be scheduled for replication anymore.
-	*/
+	/* Also remove from the replication table. */
 	hash_table_remove(m->temp_files_to_replicate, f->cached_name);
 }
 

taskvine/src/worker/vine_worker_options.c

@@ -56,7 +56,7 @@ struct vine_worker_options *vine_worker_options_create()
 	self->transfer_port_min = 0;
 	self->transfer_port_max = 0;
 
-	self->max_transfer_procs = 5;
+	self->max_transfer_procs = 10;
 
 	self->reported_transfer_host = 0;