Merge branch 'master' into task-wrapper

Author: dthain

batch_job/src/batch_job_info.c

@@ -16,5 +16,6 @@ struct batch_job_info *batch_job_info_create()
 
 void batch_job_info_delete(struct batch_job_info *info)
 {
+	free(info->schedd);
 	free(info);
 }

batch_job/src/batch_job_info.h

@@ -20,6 +20,7 @@ struct batch_job_info {
 	int exit_signal;     /**< The signal by which the job was killed, if it exited abnormally. */
 	int disk_allocation_exhausted; /**< Non-zero if the job filled its loop device allocation to capacity, zero otherwise */
 	long log_pos;        /**< Last read position in the log file, for ftell and fseek. (only for batch_queue_cluster) */
+	char *schedd;        /**< Condor schedd host from job log alias (e.g. from "Job submitted from host: <...&alias=...>"), if known. */
 };
 
 /** Create a new batch_job_info struct.

batch_job/src/batch_queue_condor.c

@@ -301,6 +301,19 @@ static batch_queue_id_t batch_queue_condor_wait(struct batch_queue *q, struct ba
 
 				if (type == 0) {
 					info->submitted = current;
+					/* Parse schedd from "Job submitted from host: <...&alias=condorfe.crc.nd.edu&...>" */
+					const char *alias = strstr(line, "alias=");
+					if (alias) {
+						alias += 6; /* skip "alias=" */
+						const char *end = strpbrk(alias, "&>");
+						if (end && end > alias) {
+							size_t len = (size_t)(end - alias);
+							free(info->schedd);
+							info->schedd = xxmalloc(len + 1);
+							memcpy(info->schedd, alias, len);
+							info->schedd[len] = '\0';
+						}
+					}
 				} else if (type == 1) {
 					info->started = current;
 					debug(D_BATCH, "job %" PRIbjid " running now", jobid);
@@ -360,7 +373,18 @@ static batch_queue_id_t batch_queue_condor_wait(struct batch_queue *q, struct ba
 
 static int batch_queue_condor_remove(struct batch_queue *q, batch_queue_id_t jobid, batch_queue_remove_mode_t mode)
 {
-	char *command = string_format("condor_rm %" PRIbjid, jobid);
+	struct batch_job_info *info = itable_lookup(q->job_table, jobid);
+	char *schedd = NULL;
+	if (info) {
+		schedd = info->schedd;
+	}
+
+	char *command = NULL;
+	if (schedd) {
+		command = string_format("condor_rm -name %s %" PRIbjid, schedd, jobid);
+	} else {
+		command = string_format("condor_rm %" PRIbjid, jobid);
+	}
 
 	debug(D_BATCH, "%s", command);
 	FILE *file = popen(command, "r");

batch_job/src/vine_factory.c

@@ -133,6 +133,65 @@ static int64_t factory_timeout = 0;
 
 static char *factory_name = NULL;
 
+/* Port range for worker-worker transfers (e.g. "10000:11000"). Passed as --transfer-port to workers. */
+static char *transfer_port_range = NULL;
+
+/* Returns 1 if valid, 0 if invalid (prints error to stderr). */
+static int validate_transfer_port_range(const char *range, const char *context)
+{
+	int min_port, max_port;
+	char *r, *ptr, *end;
+
+	if (!range || !*range) {
+		fprintf(stderr, "vine_factory: %stransfer-port must be non-empty (e.g. 10000 or 10000:11000)\n", context ? context : "");
+		return 0;
+	}
+
+	r = xxstrdup(range);
+	ptr = strtok(r, ":");
+	if (!ptr) {
+		fprintf(stderr, "vine_factory: %sinvalid transfer-port range \"%s\" (expected PORT or PORT_MIN:PORT_MAX)\n", context ? context : "", range);
+		free(r);
+		return 0;
+	}
+
+	errno = 0;
+	min_port = (int)strtol(ptr, &end, 10);
+	if (*end != '\0' || errno != 0 || min_port < 1 || min_port > 65535) {
+		fprintf(stderr, "vine_factory: %stransfer-port min must be 1-65535, got \"%s\"\n", context ? context : "", ptr);
+		free(r);
+		return 0;
+	}
+	max_port = min_port;
+
+	ptr = strtok(NULL, ":");
+	if (ptr) {
+		errno = 0;
+		max_port = (int)strtol(ptr, &end, 10);
+		if (*end != '\0' || errno != 0 || max_port < 1 || max_port > 65535) {
+			fprintf(stderr, "vine_factory: %stransfer-port max must be 1-65535, got \"%s\"\n", context ? context : "", ptr);
+			free(r);
+			return 0;
+		}
+	}
+
+	ptr = strtok(NULL, ":");
+	if (ptr) {
+		fprintf(stderr, "vine_factory: %sinvalid transfer-port range \"%s\" (expected PORT or PORT_MIN:PORT_MAX)\n", context ? context : "", range);
+		free(r);
+		return 0;
+	}
+
+	if (min_port > max_port) {
+		fprintf(stderr, "vine_factory: %stransfer-port min (%d) must not exceed max (%d)\n", context ? context : "", min_port, max_port);
+		free(r);
+		return 0;
+	}
+
+	free(r);
+	return 1;
+}
+
 struct batch_queue *queue = 0;
 
 // Whether workers should use ssl. If using the catalog server and the manager
@@ -498,11 +557,15 @@ static int submit_worker( struct batch_queue *queue )
 	if(manual_ssl_option)	ADD_ARG1(cmd," --ssl");
 	if(single_shot)		ADD_ARG1(cmd," --single-shot");
 	if(task_wrapper)        ADD_ARG2(cmd," --task-wrapper \"%s\"",task_wrapper);
+	if(transfer_port_range) ADD_ARG1(cmd," --transfer-port %s",transfer_port_range);
 	if(extra_worker_args)	ADD_ARG2(cmd," %s",extra_worker_args);
 
 	char *features_string = make_features_string(features_table);
 	ADD_ARG2(cmd," %s",features_string);
 	free(features_string);
+	if(transfer_port_range) {
+		free(transfer_port_arg);
+	}
 
 	if(wrapper_command) {
 		// Note that we don't use string_wrap_command here,
@@ -886,6 +949,7 @@ int read_config_file(const char *config_file) {
 
 	assign_new_value(new_foremen_regex, foremen_regex, foremen-name, const char *, JX_STRING, string_value)
 	assign_new_value(new_extra_worker_args, extra_worker_args, worker-extra-options, const char *, JX_STRING, string_value)
+	assign_new_value(new_transfer_port_range, transfer_port_range, transfer-port, const char *, JX_STRING, string_value)
 
 	assign_new_value(new_condor_requirements, condor_requirements, condor-requirements, const char *, JX_STRING, string_value)
 
@@ -924,6 +988,14 @@ int read_config_file(const char *config_file) {
 		error_found = 1;
 	}
 
+	if(new_transfer_port_range) {
+		char ctx[PATH_MAX + 4];
+		snprintf(ctx, sizeof(ctx), "%s: ", config_file);
+		if (!validate_transfer_port_range(new_transfer_port_range, ctx)) {
+			error_found = 1;
+		}
+	}
+
 	if(error_found) {
 		goto end;
 	}
@@ -961,6 +1033,11 @@ int read_config_file(const char *config_file) {
 		extra_worker_args = xxstrdup(new_extra_worker_args);
 	}
 
+	if(new_transfer_port_range != transfer_port_range) {
+		free(transfer_port_range);
+		transfer_port_range = new_transfer_port_range ? xxstrdup(new_transfer_port_range) : NULL;
+	}
+
 	if(new_condor_requirements != condor_requirements) {
 		free(condor_requirements);
 		condor_requirements = xxstrdup(new_condor_requirements);
@@ -1019,6 +1096,10 @@ int read_config_file(const char *config_file) {
 		fprintf(stdout, "worker-extra-options: %s", extra_worker_args);
 	}
 
+	if(transfer_port_range) {
+		fprintf(stdout, "transfer-port: %s\n", transfer_port_range);
+	}
+
 	if(workers_blocked) {
 		fprintf(stdout, "workers-blocked: %s", workers_blocked);
 	}
@@ -1073,7 +1154,7 @@ static void mainloop( struct batch_queue *queue )
 		{
 			factory_timeout_start = time(0);
 		} else {
-			// check to see if factory timeout is triggered, factory timeout will be 0 if flag isn't set
+			/* check to see if factory timeout is triggered, factory timeout will be 0 if flag isn't set */
 			if(factory_timeout > 0)
 			{
 				if(time(0) - factory_timeout_start > factory_timeout) {
@@ -1289,6 +1370,7 @@ static void show_help(const char *cmd)
 	printf("\nWorker environment options:\n");
 	printf(" %-30s Environment variable to add to worker.\n", "--env=<variable=value>");
 	printf(" %-30s Extra options to give to worker.\n", "-E,--extra-options=<options>");
+	printf(" %-30s Port range for worker-worker transfers (e.g. 10000:11000). Passed as --transfer-port.\n", "--transfer-port=<port|min:max>");
 	printf(" %-30s Alternate binary instead of vine_worker.\n", "--worker-binary=<file>");
 	printf(" %-30s Wrap worker with this command prefix.\n","--wrapper");
 	printf(" %-30s Wrap tasks with this command-prefix,\n","--task-wrapper");
@@ -1336,6 +1418,7 @@ enum{   LONG_OPT_CORES = 255,
 		LONG_OPT_DEBUG_WORKERS,
 		LONG_OPT_DISABLE_AFS_CHECK,
 		LONG_OPT_SINGLE_SHOT,
+		LONG_OPT_TRANSFER_PORT,
 };
 
 static const struct option long_options[] = {
@@ -1388,6 +1471,7 @@ static const struct option long_options[] = {
 	{"tls-sni", required_argument, 0, LONG_OPT_TLS_SNI},
 	{"factory-name",required_argument, 0, LONG_OPT_FACTORY_NAME},
 	{"single-shot", no_argument, 0, LONG_OPT_SINGLE_SHOT},
+	{"transfer-port", required_argument, 0, LONG_OPT_TRANSFER_PORT},
 	{0,0,0,0}
 };
 
@@ -1593,6 +1677,13 @@ int main(int argc, char *argv[])
 			case LONG_OPT_SINGLE_SHOT:
 				single_shot = 1;
 				break;
+			case LONG_OPT_TRANSFER_PORT:
+				if (!validate_transfer_port_range(optarg, "")) {
+					return EXIT_FAILURE;
+				}
+				free(transfer_port_range);
+				transfer_port_range = xxstrdup(optarg);
+				break;
 			default:
 				show_help(argv[0]);
 				return EXIT_FAILURE;

doc/man/m4/vine_factory.m4

@@ -98,6 +98,7 @@ OPTION_ARG_LONG(env,variable=value)
  Environment variable to add to worker.
 OPTION_ARG(E,extra-options,options)
  Extra options to give to worker.
+OPTION_ARG_LONG(transfer-port,port) Port range for worker-worker transfers (e.g. 10000:11000).
 OPTION_ARG_LONG(worker-binary,file)
  Alternate binary instead of vine_worker.
 OPTION_ARG_LONG(wrapper,cmd)
@@ -180,6 +181,7 @@ workers-per-cycle
 task-per-worker
 timeout
 worker-extra-options
+transfer-port
 condor-requirements
 cores
 memory

doc/man/md/vine_factory.md

@@ -120,6 +120,7 @@ Worker environment options:
  Environment variable to add to worker.
 - **-E**,**--extra-options=_&lt;options&gt;_**<br />
  Extra options to give to worker.
+- **--transfer-port=_&lt;port&gt;_**<br /> Port range for worker-worker transfers (e.g. 10000:11000).
 - **--worker-binary=_&lt;file&gt;_**<br />
  Alternate binary instead of vine_worker.
 - **--wrapper=_&lt;cmd&gt;_**<br />
@@ -202,6 +203,7 @@ workers-per-cycle
 task-per-worker
 timeout
 worker-extra-options
+transfer-port
 condor-requirements
 cores
 memory

doc/manuals/catalog/index.md

@@ -5,8 +5,8 @@
 Catalog servers function as connection points for tools that need to share
 information and interact remotely. Various services and tools send periodic
 updates to a catalog server to advertise their presence and vital details such
-as addresses, resources, and performance. Tools like `chirp_status` and
-`work_queue_status` query the server to displays servers that are currently
+as addresses, resources, and performance. Tools like `vine_status`, `work_queue_status`, and `chirp_status` and
+query the catalog to displays service that are currently
 running. Catalog updates are sent via TCP or UDP, and the catalog server exposes a
 JSON interface to view status and make queries.
 
@@ -19,11 +19,13 @@ automatic backup) at Notre Dame:
 
 The default view for a catalog server is a human-readable HTML summary.
 Machine-readable data is also available as JSON, text, XML, or ClassAds. Many
-parts of cctools make use of a catalog server internally. Chirp servers send
+parts of cctools make use of a catalog server internally. [Chirp](../chirp) servers send
 regular catalog updates indicating the host system's load, available disk
-space, cctools version, etc. Work Queue managers also advertise their projects
-through the catalog. When a worker starts, it can query the catalog to
-automatically discover a manager to contact.
+space, cctools version, etc.
+[TaskVine](../taskvine) and [Work Queue](../workqueue) managers advertise
+currently running applications, indicating number of tasks, workers, performance, etc.
+When workers start, they can query the catalog to
+automatically discover which manager to contact.
 
 ## Specifying Catalog Servers
 
@@ -33,15 +35,14 @@ delimited list of servers to use. Each may optionally include a port number.
 If no port is specified, the value of the environment variable `CATALOG_PORT`
 is used, or the default of port 9097. If no catalog server is given on the
 command line, the `CATALOG_HOST` environment variable is used. If that is
-unset, the default of `catalog.cse.nd.edu,backup-catalog.cse.nd.edu` This
-could be written more verbosely as `catalog.cse.nd.edu:9097,backup-catalog.cse.nd.edu:9097` assuming the catalog port was not set in the
-environment.
+unset, the default of `catalog.cse.nd.edu,backup-catalog.cse.nd.edu` is used,
+assuming a default port of 9097.
 
 ## Querying Catalog Servers
 
 There are several ways to query a catalog server. If you are querying
-specifically for Chirp servers or Work Queue applications, then use the
-`chirp_status` or `work_queue_status` tools, which query the server and
+specifically for Chirp servers or TaskVine applications, then use the
+`chirp_status` or `vine_status` tools, which query the server and
 display fields specific for those uses.
 
 To view all kinds of records in raw JSON format, use the `catalog_query` tool.
@@ -204,5 +205,5 @@ For more information, please see [Getting Help](../help.md) or visit the [Cooper
 
 ## Copyright
 
-CCTools is Copyright (C) 2022 The University of Notre Dame. This software is distributed under the GNU General Public License Version 2. See the file COPYING for
+CCTools is Copyright (C) 2026 The University of Notre Dame. This software is distributed under the GNU General Public License Version 2. See the file COPYING for
 details.

dttools/src/debug.c

@@ -115,6 +115,14 @@ struct fatal_callback {
 
 struct fatal_callback *fatal_callback_list = 0;
 
+static void debug_signal_handler(int sig)
+{
+	signal(sig, SIG_DFL);
+	debug(D_ERROR, "Fatal error: signal %d received", sig);
+	debug_backtrace;
+	_exit(1);
+}
+
 int debug_flags_set(const char *flagname)
 {
 	struct flag_info *i;
@@ -280,6 +288,8 @@ void fatal(const char *fmt, ...)
 	struct fatal_callback *f;
 	va_list args;
 
+	debug_backtrace;
+
 	va_start(args, fmt);
 	do_debug(D_FATAL, fmt, args);
 	va_end(args);
@@ -323,6 +333,15 @@ void debug_config_file(const char *path)
 		fprintf(stderr, "could not set debug file '%s': %s", path, strerror(errno));
 		exit(EXIT_FAILURE);
 	}
+
+	char *var = getenv("CCTOOLS_DEBUG_CAPTURE_SIGNALS");
+	if (var) {
+		signal(SIGSEGV, debug_signal_handler);
+		signal(SIGILL, debug_signal_handler);
+		signal(SIGTERM, debug_signal_handler);
+		signal(SIGINT, debug_signal_handler);
+		signal(SIGQUIT, debug_signal_handler);
+	}
 }
 
 void debug_config(const char *name)

dttools/src/debug.h

@@ -35,6 +35,7 @@ unless it has the flags D_NOTICE or D_FATAL.  For example, a main program might
 #include "int_sizes.h"
 
 #include <unistd.h>
+#include <execinfo.h>
 
 #include <sys/types.h>
 
@@ -264,4 +265,30 @@ void debug_close(void);
  */
 #define LDEBUG(fmt, ...) debug(D_DEBUG, "%s:%s:%d[%s]: " fmt, __func__, __FILE__, __LINE__, CCTOOLS_SOURCE, __VA_ARGS__)
 
+
+#define DEBUG_BT_SIZE 100
+
+/* Use offset info (+0x...) with addr2line -f -e [exe or .so] */
+#define debug_backtrace \
+	{ debug(D_ERROR, "%s:%s:%d[%s]", __func__, __FILE__, __LINE__, CCTOOLS_SOURCE); \
+		void *buffer[DEBUG_BT_SIZE]; \
+		size_t nptrs = backtrace(buffer, DEBUG_BT_SIZE); \
+		char **strings = backtrace_symbols(buffer, nptrs); \
+		if (!strings) { \
+			debug(D_ERROR, "%s", "No backtrace available."); \
+		} else { \
+			for (size_t bt_ct = 0; bt_ct < nptrs; bt_ct++) { \
+				debug(D_ERROR, "%s", strings[bt_ct]); \
+			} \
+		} \
+		free(strings); \
+	} // generate a core if possible
+
+
+#define debug_assert(cond)\
+  if (!(cond)) {\
+    debug(D_ERROR, "Assertion failed: %s", #cond);\
+    debug_backtrace;\
+    abort(); }
+
 #endif