Batch: Rework Interface Into Queue, Task, File Objects (#3898)

* Rework batch interface to accept batch_job_submit(struct batch_task *t).
This will allow us to pass properly formed file objects and make use of
more features of taskvine from makeflow and the factory.

* Makeflow now uses batch_job_submit(queue,task) as promised long ago.

* Blue Waters was decomissioned in 2023.

* Remove largely unused batch_fs operations.

* Clean up batch_queue structures.

* Revert batch_fs operations back to basic unix operations.

* Remove unused functions, format for consistency.

* Separate batch_job_info into a separate file.

* Use batch_queue consistently to describe a queue.

* batch_task -> batch_job

* Update copyright boilerplate

* Update batch_queue_cluster.c to new interface.

* Update batch_queue_amazon.c to new interface.

* Add back in batch_queue_amazon and batch_queue_cluster.

* Replace throughout:
batch_job -> batch_queue
batch_task -> batch_job

* Remove mesos from work_queue_factory.

* Remove mesos support.

* Remove MPI from makeflow.

* Remove unused makeflow-mpi support.

* Fix up batch_queue_k8s by transforming file lists to strings in batch_queue_k8s_submit.

* Fix incorrect ordering of inner file / outer file.

* batch_fs_unlink is replaced by unlink_recursive, not just unlink (caught by TR_makeflow_restart.sh)

* Produce a factory log in test.

* Do not attach worker log as output unless debugging is requested.

* Move retired modules to old

* Clean up makefile to incorporate new headers.

* Complete autodocs for batch_job/file/queue/wrapper

* Remove amazon_batch, lambda, and mesos options from makeflow.

* Remove Amazon Batch, Lambda, Mesos, and MPI from man pages.

* Remove Amazon Batch, Lambda, Mesos, MPI from manual.

* explicitely -> explicitly

* Remove unused chirp symbol

* Make autodocs example complete with input file.

* Remove unused items from makeflow man page.

* Reorder and doc batch system types.

* General renaming of sge to uge in code and manuals.
sge is retained as an alias for uge in batch queue selection.

* sge -> uge in vine_submit_workers

* sge -> uge

* format and lint batch_job module

* Apply clang formatting rules.

* Add "experimental" message for batch modules in which we have lower confidence.

* queue -> remote_queue

* Fix up condor afs error message

* experimental flag should be a feature not an option

Author: dthain

Makefile

@@ -49,13 +49,13 @@ install: $(INSTALL_PACKAGES)
 test: $(CCTOOLS_PACKAGES)
 	./run_all_tests.sh
 
-PACKAGES_TO_LINT = taskvine dttools poncho resource_monitor
+PACKAGES_TO_LINT = taskvine resource_monitor batch_job dttools poncho
 LINT_PACKAGES = $(PACKAGES_TO_LINT:%=lint-%)
 $(LINT_PACKAGES): config.mk
 	@$(MAKE) -C $(@:lint-%=%) lint
 lint: $(LINT_PACKAGES)
 
-PACKAGES_TO_FORMAT = taskvine resource_monitor dttools
+PACKAGES_TO_FORMAT = taskvine resource_monitor batch_job dttools
 FORMAT_PACKAGES = $(PACKAGES_TO_FORMAT:%=format-%)
 $(FORMAT_PACKAGES): config.mk
 	@$(MAKE) -C $(@:format-%=%) format

batch_job/src/Makefile

@@ -13,28 +13,21 @@ VINE_LIB=${CCTOOLS_HOME}/taskvine/src/manager/libtaskvine.a
 EXTERNAL_LIBRARIES=$(CHIRP_LIB) $(DT_LIB)
 
 SOURCES = \
+	batch_queue.c \
+	batch_job.c \
+	batch_job_info.c \
 	batch_file.c \
-	batch_task.c \
 	batch_wrapper.c \
-	batch_job.c \
-	batch_job_amazon.c \
-	batch_job_amazon_batch.c \
-	batch_job_dryrun.c \
-	$(CHIRP_BATCH) \
-	batch_job_cluster.c \
-	batch_job_blue_waters.c \
-	batch_job_condor.c \
-	batch_job_local.c \
-	batch_job_work_queue.c\
-	batch_job_lambda.c \
-	batch_job_work_queue.c \
-	batch_job_vine.c \
-	batch_job_mesos.c \
-	batch_job_k8s.c \
-	batch_job_mpi.c \
-	mesos_task.c
-
-PUBLIC_HEADERS = batch_job.h
+	batch_queue_local.c \
+	batch_queue_dryrun.c \
+	batch_queue_condor.c \
+	batch_queue_vine.c \
+	batch_queue_work_queue.c \
+	batch_queue_cluster.c \
+	batch_queue_k8s.c \
+	batch_queue_amazon.c
+
+PUBLIC_HEADERS = batch_queue.h batch_job.h batch_job_info.h batch_file.h batch_wrapper.h
 
 OBJECTS = $(SOURCES:%.c=%.o)
 
@@ -61,5 +54,17 @@ install:
 	cp $(PUBLIC_HEADERS) $(CCTOOLS_INSTALL_DIR)/include/cctools
 
 clean:
-	rm -rf $(OBJECTS) $(LIBRARIES) $(PROGRAMS) batch_job_amazon_script.c *.o
+	rm -rf $(OBJECTS) $(LIBRARIES) $(PROGRAMS) *.o
+
+lint:
+	if ( ! clang-format -Werror --dry-run --style='file:../../.clang-format' $(SOURCES));\
+	then\
+		echo  "========================================================";\
+		echo  "NOTICE: Run `make format` to format your latest changes.";\
+		echo  "========================================================";\
+		exit 1;\
+	fi
+
+format:
+	clang-format -i $(SOURCES)
 

batch_job/src/batch_file.c

@@ -1,15 +1,19 @@
 /*
-Copyright (C) 2022 The University of Notre Dame
+Copyright (C) 2024 The University of Notre Dame
 This software is distributed under the GNU General Public License.
 See the file COPYING for details.
 */
 
+#include "batch_queue.h"
+#include "batch_job.h"
 #include "batch_file.h"
+
 #include "sha1.h"
 #include "stringtools.h"
 #include "xxmalloc.h"
 #include "path.h"
 #include "hash_table.h"
+
 #include <time.h>
 #include <sys/time.h>
 #include <sys/types.h>
@@ -26,26 +30,24 @@ double total_checksum_time = 0.0;
  *  IF no inner_name is given, or the specified batch_queue does not support
  *  remote renaming the outer_name will be used.
  **/
-struct batch_file *batch_file_create(struct batch_queue *queue, const char * outer_name, const char * inner_name)
+struct batch_file *batch_file_create(const char *outer_name, const char *inner_name)
 {
-	struct batch_file *f = calloc(1,sizeof(*f));
-    f->outer_name = xxstrdup(outer_name);
-
-	if(batch_queue_supports_feature(queue, "remote_rename") && inner_name){
+	struct batch_file *f = calloc(1, sizeof(*f));
+	f->outer_name = xxstrdup(outer_name);
+	if (inner_name) {
 		f->inner_name = xxstrdup(inner_name);
 	} else {
 		f->inner_name = xxstrdup(outer_name);
 	}
-
-    return f;
+	return f;
 }
 
 /**
  * Delete batch_file, including freeing outer_name and inner_name/
  **/
 void batch_file_delete(struct batch_file *f)
 {
-	if(!f)
+	if (!f)
 		return;
 
 	free(f->outer_name);
@@ -54,135 +56,123 @@ void batch_file_delete(struct batch_file *f)
 	free(f);
 }
 
-/**
- * Given a file, return the string that identifies it appropriately
- * for the given batch system, combining the local and remote name
- * and making substitutions according to the node.
- **/
-char * batch_file_to_string(struct batch_queue *queue, struct batch_file *f )
+char *batch_file_to_string(struct batch_file *f)
 {
-    if(batch_queue_supports_feature(queue,"remote_rename")) {
-            return string_format("%s=%s", f->outer_name, f->inner_name);
-    } else {
-            return string_format("%s", f->outer_name);
-    }
+	if (!strcmp(f->inner_name, f->outer_name)) {
+		return strdup(f->outer_name);
+	} else {
+		return string_format("%s=%s", f->outer_name, f->inner_name);
+	}
 }
 
-/**
- * Given a list of files, add the files to the given string.
- * Returns the original string, realloced if necessary
- **/
-char * batch_files_to_string(struct batch_queue *queue, struct list *files )
+char *batch_file_list_to_string(struct list *file_list)
 {
-    struct batch_file *file;
-
-    char * file_str = strdup("");
+	struct batch_file *file;
 
-	char * separator = "";
+	char *file_str = strdup("");
+	char *separator = "";
 
-    if(!files) return file_str;
+	if (!file_list)
+		return file_str;
 
-    list_first_item(files);
-    while((file=list_next_item(files))) {
+	LIST_ITERATE(file_list, file)
+	{
 		/* Only add separator if past first item. */
-		file_str = string_combine(file_str,separator);
+		file_str = string_combine(file_str, separator);
 
-        char *f = batch_file_to_string(queue, file);
-        file_str = string_combine(file_str,f);
-	
-		/* This could be set using batch_queue feature or option 
+		char *f = batch_file_to_string(file);
+		file_str = string_combine(file_str, f);
+
+		/* This could be set using batch_queue feature or option
 		 * to allow for batch system specific separators. */
 		separator = ",";
 
-        free(f);
-    }
+		free(f);
+	}
 
-    return file_str;
+	return file_str;
 }
 
-int batch_file_outer_compare(const void *file1, const void *file2) {
-	struct batch_file **f1 = (void *)file1;
-	struct batch_file **f2 = (void *)file2;
-
-	return strcmp((*f1)->outer_name, (*f2)->outer_name);
+int batch_file_outer_compare(struct batch_file *file1, struct batch_file *file2)
+{
+	return strcmp(file1->outer_name, file2->outer_name);
 }
 
 /* Return the content based ID for a file.
  * generates the checksum of a file's contents if does not exist */
-char * batch_file_generate_id(struct batch_file *f) {
-	if(check_sums == NULL){
-		check_sums = hash_table_create(0,0);
-		}
+char *batch_file_generate_id(struct batch_file *f)
+{
+	if (check_sums == NULL) {
+		check_sums = hash_table_create(0, 0);
+	}
 	char *check_sum_value = hash_table_lookup(check_sums, f->outer_name);
-	if(check_sum_value == NULL){
+	if (check_sum_value == NULL) {
 		unsigned char hash[SHA1_DIGEST_LENGTH];
 		struct timeval start_time;
-			struct timeval end_time;
+		struct timeval end_time;
 
-		gettimeofday(&start_time,NULL);
+		gettimeofday(&start_time, NULL);
 		int success = sha1_file(f->outer_name, hash);
-		gettimeofday(&end_time,NULL);
-			double run_time = ((end_time.tv_sec*1000000 + end_time.tv_usec) - (start_time.tv_sec*1000000 + start_time.tv_usec)) / 1000000.0;
-			total_checksum_time += run_time;
-			debug(D_MAKEFLOW_HOOK," The total checksum time is %lf",total_checksum_time);
-		if(success == 0){
+		gettimeofday(&end_time, NULL);
+		double run_time = ((end_time.tv_sec * 1000000 + end_time.tv_usec) - (start_time.tv_sec * 1000000 + start_time.tv_usec)) / 1000000.0;
+		total_checksum_time += run_time;
+		debug(D_MAKEFLOW_HOOK, " The total checksum time is %lf", total_checksum_time);
+		if (success == 0) {
 			debug(D_MAKEFLOW, "Unable to checksum this file: %s", f->outer_name);
 			return NULL;
 		}
 		f->hash = xxstrdup(sha1_string(hash));
 		hash_table_insert(check_sums, f->outer_name, xxstrdup(sha1_string(hash)));
-		debug(D_MAKEFLOW,"Checksum hash of %s is: %s",f->outer_name,f->hash);
+		debug(D_MAKEFLOW, "Checksum hash of %s is: %s", f->outer_name, f->hash);
 		return xxstrdup(f->hash);
 	}
-	debug(D_MAKEFLOW,"Checksum already exists in hash table. Cached CHECKSUM hash of %s is: %s", f->outer_name, check_sum_value);
+	debug(D_MAKEFLOW, "Checksum already exists in hash table. Cached CHECKSUM hash of %s is: %s", f->outer_name, check_sum_value);
 	return xxstrdup(check_sum_value);
 }
 
-
 /* Return the content based ID for a directory.
  * generates the checksum for the directories contents if does not exist
  * 		*NEED TO ACCOUNT FOR SYMLINKS LATER*  */
-char *  batch_file_generate_id_dir(char *file_name){
-	if(check_sums == NULL){
-		check_sums = hash_table_create(0,0);
+char *batch_file_generate_id_dir(char *file_name)
+{
+	if (check_sums == NULL) {
+		check_sums = hash_table_create(0, 0);
 	}
 	char *check_sum_value = hash_table_lookup(check_sums, file_name);
-	if(check_sum_value == NULL){
+	if (check_sum_value == NULL) {
 		char *hash_sum = "";
 		struct dirent **dp;
 		int num;
 		// Scans directory and sorts in reverse order
 		num = scandir(file_name, &dp, NULL, alphasort);
-		if(num < 0){
-			debug(D_MAKEFLOW,"Unable to scan %s", file_name);
+		if (num < 0) {
+			debug(D_MAKEFLOW, "Unable to scan %s", file_name);
 			return NULL;
-		}
-		else{
+		} else {
 			int i;
-			for(i = num - 1; i >= 0; i--) {
-				if(strcmp(dp[i]->d_name,".") != 0 && strcmp(dp[i]->d_name,"..") != 0){
-					char *file_path = string_format("%s/%s",file_name,dp[i]->d_name);
-					if(path_is_dir(file_path) == 1){
-						hash_sum = string_format("%s%s",hash_sum,batch_file_generate_id_dir(file_path));
-					}
-					else{
+			for (i = num - 1; i >= 0; i--) {
+				if (strcmp(dp[i]->d_name, ".") != 0 && strcmp(dp[i]->d_name, "..") != 0) {
+					char *file_path = string_format("%s/%s", file_name, dp[i]->d_name);
+					if (path_is_dir(file_path) == 1) {
+						hash_sum = string_format("%s%s", hash_sum, batch_file_generate_id_dir(file_path));
+					} else {
 						unsigned char hash[SHA1_DIGEST_LENGTH];
 						struct timeval start_time;
-							struct timeval end_time;
+						struct timeval end_time;
 
-						gettimeofday(&start_time,NULL);
+						gettimeofday(&start_time, NULL);
 						int success = sha1_file(file_path, hash);
-						gettimeofday(&end_time,NULL);
-							double run_time = ((end_time.tv_sec*1000000 + end_time.tv_usec) - (start_time.tv_sec*1000000 + start_time.tv_usec)) / 1000000.0;
-							total_checksum_time += run_time;
-							debug(D_MAKEFLOW_HOOK," The total checksum time is %lf",total_checksum_time);
-						if(success == 0){
+						gettimeofday(&end_time, NULL);
+						double run_time = ((end_time.tv_sec * 1000000 + end_time.tv_usec) - (start_time.tv_sec * 1000000 + start_time.tv_usec)) / 1000000.0;
+						total_checksum_time += run_time;
+						debug(D_MAKEFLOW_HOOK, " The total checksum time is %lf", total_checksum_time);
+						if (success == 0) {
 							debug(D_MAKEFLOW, "Unable to checksum this file: %s", file_path);
 							free(file_path);
 							free(dp[i]);
 							continue;
 						}
-						hash_sum = string_format("%s%s:%s",hash_sum,file_name,sha1_string(hash));
+						hash_sum = string_format("%s%s:%s", hash_sum, file_name, sha1_string(hash));
 					}
 					free(file_path);
 				}
@@ -193,10 +183,10 @@ char *  batch_file_generate_id_dir(char *file_name){
 			sha1_buffer(hash_sum, strlen(hash_sum), hash);
 			free(hash_sum);
 			hash_table_insert(check_sums, file_name, xxstrdup(sha1_string(hash)));
-			debug(D_MAKEFLOW,"Checksum hash of %s is: %s",file_name,sha1_string(hash));
+			debug(D_MAKEFLOW, "Checksum hash of %s is: %s", file_name, sha1_string(hash));
 			return xxstrdup(sha1_string(hash));
 		}
 	}
-	debug(D_MAKEFLOW,"Checksum already exists in hash table. Cached CHECKSUM hash of %s is: %s", file_name, check_sum_value);
+	debug(D_MAKEFLOW, "Checksum already exists in hash table. Cached CHECKSUM hash of %s is: %s", file_name, check_sum_value);
 	return check_sum_value;
 }