add compute makespan

Author: Jin Zhou

taskvine/src/bindings/python3/ndcctools/taskvine/compat/dask_executor.py

@@ -138,8 +138,10 @@ def get(self, dsk, keys, *,
             hoisting_modules=None,  # Deprecated, use lib_modules
             import_modules=None,    # Deprecated, use lib_modules
             lazy_transfers=True,    # Deprecated, use worker_tranfers
+            extra_serialize_time_sec=0,
             ):
         try:
+            self.extra_serialize_time_sec = extra_serialize_time_sec
             self.set_property("framework", "dask")
             if retries and retries < 1:
                 raise ValueError("retries should be larger than 0")
@@ -213,10 +215,13 @@ def __call__(self, *args, **kwargs):
         return self.get(*args, **kwargs)
 
     def _dask_execute(self, dsk, keys):
+
         indices = {k: inds for (k, inds) in find_dask_keys(keys)}
         keys_flatten = indices.keys()
 
+        time_start = time.time()
         dag = DaskVineDag(dsk, low_memory_mode=self.low_memory_mode, prune_depth=self.prune_depth)
+        print(f"Time taken to enqueue tasks: {time.time() - time_start:.6f} seconds")
         tag = f"dag-{id(dag)}"
 
         # create Library if using 'function-calls' task mode.
@@ -437,7 +442,8 @@ def _enqueue_dask_calls(self, dag, tag, rs, retries, enqueued_calls):
                                      extra_files=self.extra_files,
                                      retries=retries,
                                      worker_transfers=lazy,
-                                     wrapper=self.wrapper)
+                                     wrapper=self.wrapper,
+                                     extra_serialize_time_sec=self.extra_serialize_time_sec)
 
                 t.set_priority(priority)
                 t.set_tag(tag)  # tag that identifies this dag
@@ -557,7 +563,9 @@ def __init__(self, m,
                  env_vars=None,
                  retries=5,
                  worker_transfers=False,
-                 wrapper=None):
+                 wrapper=None,
+                 extra_serialize_time_sec=0):
+        time.sleep(extra_serialize_time_sec)
         self._key = key
         self._sexpr = sexpr
 
@@ -658,7 +666,9 @@ def __init__(self, m,
                  extra_files=None,
                  retries=5,
                  worker_transfers=False,
-                 wrapper=None):
+                 wrapper=None,
+                 extra_serialize_time_sec=0):
+        time.sleep(extra_serialize_time_sec)
 
         self._key = key
         self.resources = resources

taskvine/src/graph/dagvine/dagvine.py

@@ -16,6 +16,7 @@
 import signal
 import json
 import random
+import time
 
 
 def context_loader_func(graph_pkl):
@@ -69,7 +70,7 @@ def __init__(self):
             "checkpoint-dir": "./checkpoints",
             "checkpoint-fraction": 0,
             "progress-bar-update-interval-sec": 0.1,
-            "time-metrics-filename": "time_metrics.csv",
+            "time-metrics-filename": 0,
             "enable-debug-log": 1,
             "auto-recovery": 1,
             "max-retry-attempts": 15,
@@ -258,6 +259,8 @@ def create_proxy_library(self, py_graph, vine_graph, hoisting_modules, env_files
 
     def run(self, task_dict, target_keys=[], params={}, hoisting_modules=[], env_files={}, adapt_dask=False):
         """High-level entry point: normalise input, build graphs, ship the library, execute, and return results."""
+        time_start = time.time()
+
         # first update the params so that they can be used for the following construction
         self.update_params(params)
 
@@ -278,13 +281,18 @@ def run(self, task_dict, target_keys=[], params={}, hoisting_modules=[], env_fil
 
         try:
             print(f"=== Library serialized size: {color_text(proxy_library.get_context_size(), 92)} MB")
+            print(f"Time taken to initialize the graph in Python: {time.time() - time_start:.6f} seconds")
             vine_graph.execute()
             results = {}
             for k in target_keys:
                 if k not in py_graph.task_dict:
                     continue
                 outfile_path = os.path.join(self.param("output-dir"), py_graph.outfile_remote_name[k])
                 results[k] = TaskOutputWrapper.load_from_path(outfile_path)
+            makespan_s = round(vine_graph.get_makespan_us() / 1e6, 6)
+            throughput_tps = round(len(py_graph.task_dict) / makespan_s, 6)
+            print(f"Makespan: {color_text(makespan_s, 92)} seconds")
+            print(f"Throughput: {color_text(throughput_tps, 92)} tasks/second")
             return results
         finally:
             try:

taskvine/src/graph/dagvine/vine_graph/vine_graph.c

@@ -563,13 +563,13 @@ static void print_time_metrics(struct vine_graph *vg, const char *filename)
 		debug(D_ERROR, "failed to open file %s", filename);
 		return;
 	}
-	fprintf(fp, "node_id,submission_time_us,scheduling_time_us,commit_time_us,execution_time_us,retrieval_time_us,postprocessing_time_us\n");
+	fprintf(fp, "node_id,submission_time_us,commit_time_us,execution_time_us,retrieval_time_us,postprocessing_time_us\n");
 
 	uint64_t nid;
 	struct vine_node *node;
 	ITABLE_ITERATE(vg->nodes, nid, node)
 	{
-		fprintf(fp, "%" PRIu64 "," TIMESTAMP_FORMAT "," TIMESTAMP_FORMAT "," TIMESTAMP_FORMAT "," TIMESTAMP_FORMAT "," TIMESTAMP_FORMAT "," TIMESTAMP_FORMAT "\n", node->node_id, node->submission_time, node->scheduling_time, node->commit_time, node->execution_time, node->retrieval_time, node->postprocessing_time);
+		fprintf(fp, "%" PRIu64 "," TIMESTAMP_FORMAT "," TIMESTAMP_FORMAT "," TIMESTAMP_FORMAT "," TIMESTAMP_FORMAT "," TIMESTAMP_FORMAT "\n", node->node_id, node->submission_time, node->commit_time, node->execution_time, node->retrieval_time, node->postprocessing_time);
 	}
 	fclose(fp);
 
@@ -738,7 +738,7 @@ int vine_graph_tune(struct vine_graph *vg, const char *name, const char *value)
 		vg->progress_bar_update_interval_sec = (val > 0.0) ? val : 0.1;
 
 	} else if (strcmp(name, "time-metrics-filename") == 0) {
-		if (strcmp(value, "0") == 0) {
+		if (value == NULL || strcmp(value, "0") == 0) {
 			return 0;
 		}
 
@@ -1255,6 +1255,10 @@ struct vine_graph *vine_graph_create(struct vine_manager *q)
 	 * we can be in control of when to recreate what lost data. */
 	vg->auto_recovery = 0;
 
+	vg->time_first_task_dispatched = UINT64_MAX;
+	vg->time_last_task_retrieved = 0;
+	vg->makespan_us = 0;
+
 	return vg;
 }
 
@@ -1295,6 +1299,15 @@ void vine_graph_add_dependency(struct vine_graph *vg, uint64_t parent_id, uint64
 	return;
 }
 
+uint64_t vine_graph_get_makespan_us(const struct vine_graph *vg)
+{
+	if (!vg) {
+		return 0;
+	}
+
+	return (uint64_t)vg->makespan_us;
+}
+
 /**
  * Execute the vine graph. This must be called after all nodes and dependencies are added and the topology metrics are computed.
  * @param vg Reference to the vine graph.
@@ -1305,6 +1318,8 @@ void vine_graph_execute(struct vine_graph *vg)
 		return;
 	}
 
+	timestamp_t time_start = timestamp_get();
+
 	void (*previous_sigint_handler)(int) = signal(SIGINT, handle_sigint);
 
 	debug(D_VINE, "start executing vine graph");
@@ -1363,6 +1378,9 @@ void vine_graph_execute(struct vine_graph *vg)
 		next_failure_threshold = vg->failure_injection_step_percent / 100.0;
 	}
 
+	timestamp_t time_end = timestamp_get();
+	printf("Time taken to initialize the graph in C: %.6f seconds\n", (double)(time_end - time_start) / 1e6);
+
 	struct ProgressBar *pbar = progress_bar_init("Executing Tasks");
 	progress_bar_set_update_interval(pbar, vg->progress_bar_update_interval_sec);
 
@@ -1425,6 +1443,15 @@ void vine_graph_execute(struct vine_graph *vg)
 				continue;
 			}
 
+			/* update time metrics */
+			vg->time_first_task_dispatched = MIN(vg->time_first_task_dispatched, task->time_when_commit_end);
+			vg->time_last_task_retrieved = MAX(vg->time_last_task_retrieved, task->time_when_retrieval);
+			if (vg->time_last_task_retrieved < vg->time_first_task_dispatched) {
+				debug(D_ERROR, "task %d time_last_task_retrieved < time_first_task_dispatched: %" PRIu64 " < %" PRIu64, task->task_id, vg->time_last_task_retrieved, vg->time_first_task_dispatched);
+				vg->time_last_task_retrieved = vg->time_first_task_dispatched;
+			}
+			vg->makespan_us = vg->time_last_task_retrieved - vg->time_first_task_dispatched;
+
 			/* if the outfile is set to save on the sharedfs, stat to get the size of the file */
 			switch (node->outfile_type) {
 			case NODE_OUTFILE_TYPE_SHARED_FILE_SYSTEM: {
@@ -1450,10 +1477,9 @@ void vine_graph_execute(struct vine_graph *vg)
 			 * Only the first completion should advance the "User" progress. */
 			int first_completion = !node->completed;
 			node->completed = 1;
-			node->scheduling_time = task->time_when_scheduling_end - task->time_when_scheduling_start;
 			node->commit_time = task->time_when_commit_end - task->time_when_commit_start;
 			node->execution_time = task->time_workers_execute_last;
-			node->retrieval_time = task->time_when_get_result_end - task->time_when_get_result_start;
+			node->retrieval_time = task->time_when_done - task->time_when_retrieval;
 
 			/* prune nodes on task completion */
 			prune_ancestors_of_node(vg, node);

taskvine/src/graph/dagvine/vine_graph/vine_graph.h

@@ -11,6 +11,7 @@
 #include "set.h"
 #include "vine_node.h"
 #include "taskvine.h"
+#include "timestamp.h"
 
 /** The task priority algorithm used for vine graph scheduling. */
 typedef enum {
@@ -72,6 +73,10 @@ struct vine_graph {
 
 	int max_retry_attempts;	   /* the maximum number of times to retry a task */
 	double retry_interval_sec; /* the interval between retries in seconds, 0 means no retry interval */
+
+	timestamp_t time_first_task_dispatched; /* the time when the first task is dispatched */
+	timestamp_t time_last_task_retrieved;	/* the time when the last task is retrieved */
+	timestamp_t makespan_us;		/* the makespan of the vine graph in microseconds */
 };
 
 /* Public APIs for operating the vine graph */
@@ -171,4 +176,10 @@ void vine_graph_set_proxy_function_name(struct vine_graph *vg, const char *proxy
 */
 int vine_graph_tune(struct vine_graph *vg, const char *name, const char *value);
 
+/** Get the makespan of the vine graph in microseconds.
+@param vg Reference to the vine graph.
+@return The makespan in microseconds.
+*/
+uint64_t vine_graph_get_makespan_us(const struct vine_graph *vg);
+
 #endif // VINE_GRAPH_H

taskvine/src/graph/dagvine/vine_graph/vine_graph_client.py

@@ -84,6 +84,10 @@ def execute(self):
         """Kick off execution; runs through SWIG down into the C orchestration loop."""
         vine_graph_capi.vine_graph_execute(self._c_graph)
 
+    def get_makespan_us(self):
+        """Get the makespan of the vine graph in microseconds."""
+        return vine_graph_capi.vine_graph_get_makespan_us(self._c_graph)
+
     def delete(self):
         """Release the C resources and clear the client."""
         vine_graph_capi.vine_graph_delete(self._c_graph)

taskvine/src/manager/vine_manager.c

@@ -595,8 +595,6 @@ static vine_result_code_t get_completion_result(struct vine_manager *q, struct v
 		return VINE_SUCCESS;
 	}
 
-	t->time_when_get_result_start = timestamp_get();
-
 	if (task_status != VINE_RESULT_SUCCESS) {
 		w->last_failure_time = timestamp_get();
 		t->time_when_last_failure = w->last_failure_time;

taskvine/src/manager/vine_task.c

@@ -89,15 +89,9 @@ struct vine_task *vine_task_create(const char *command_line)
 
 void vine_task_clean(struct vine_task *t)
 {
-	t->time_when_scheduling_start = 0;
-	t->time_when_scheduling_end = 0;
-
 	t->time_when_commit_start = 0;
 	t->time_when_commit_end = 0;
 
-	t->time_when_get_result_start = 0;
-	t->time_when_get_result_end = 0;
-
 	t->time_when_retrieval = 0;
 	t->time_when_done = 0;
 
@@ -161,15 +155,9 @@ void vine_task_reset(struct vine_task *t)
 	t->time_workers_execute_exhaustion = 0;
 	t->time_workers_execute_failure = 0;
 
-	t->time_when_scheduling_start = 0;
-	t->time_when_scheduling_end = 0;
-
 	t->time_when_commit_start = 0;
 	t->time_when_commit_end = 0;
 
-	t->time_when_get_result_start = 0;
-	t->time_when_get_result_end = 0;
-
 	rmsummary_delete(t->resources_measured);
 	rmsummary_delete(t->resources_allocated);
 	t->resources_measured = rmsummary_create(-1);

taskvine/src/manager/vine_task.h

@@ -117,20 +117,13 @@ struct vine_task {
 	timestamp_t time_when_submitted;    /**< The time at which this task was added to the queue. */
 	timestamp_t time_when_done;         /**< The time at which the task is mark as retrieved, after transfering output files and other final processing. */
 
-	timestamp_t time_when_scheduling_start; /**< The time when the task starts to be considered for scheduling. */
-	timestamp_t time_when_scheduling_end;   /**< The time when the task is mapped to a worker and ready to be committed. */
-
 	timestamp_t time_when_commit_start; /**< The time when the task starts to be transfered to a worker. */
 	timestamp_t time_when_commit_end;   /**< The time when the task is completely transfered to a worker. */
 
-	timestamp_t time_when_get_result_start; /**< The time when the task starts to get the result from the worker. */
-	timestamp_t time_when_get_result_end;   /**< The time when the task gets the result from the worker. */
-
 	timestamp_t time_when_retrieval;    /**< The time when output files start to be transfered back to the manager. time_done - time_when_retrieval is the time taken to transfer output files. */
 
 	timestamp_t time_when_last_failure; /**< If larger than 0, the time at which the last task failure was detected. */
 
-
 	timestamp_t time_workers_execute_last_start;           /**< The time when the last complete execution for this task started at a worker. */
 	timestamp_t time_workers_execute_last_end;             /**< The time when the last complete execution for this task ended at a worker. */