add worker code

Author: colinthomas-z80

taskvine/src/manager/vine_manager.c

@@ -2724,41 +2724,6 @@ static vine_result_code_t start_one_task(struct vine_manager *q, struct vine_wor
 	return result;
 }
 
-/*
-Start one task on a given worker by specializing the task to the worker,
-sending the appropriate input files, and then sending the details of the task.
-Note that the "infile" and "outfile" components of the task refer to
-files that have already been uploaded into the worker's cache by the manager.
-*/
-
-static vine_result_code_t start_group_task(
-		struct vine_manager *q, struct vine_worker_info *w, struct vine_task *t, struct list *l)
-{
-	vine_result_code_t result = 0;
-	struct vine_task *lt;
-	LIST_ITERATE(l, lt)
-	{
-		struct rmsummary *limits = vine_manager_choose_resources_for_task(q, w, t);
-		char *command_line;
-
-		if (q->monitor_mode && !t->needs_library) {
-			command_line = vine_monitor_wrap(q, w, t, limits);
-		} else {
-			command_line = xxstrdup(t->command_line);
-		}
-		result = vine_manager_put_task(q, w, lt, command_line, limits, 0);
-		free(command_line);
-		if (result == VINE_SUCCESS) {
-			t->current_resource_box = limits;
-			rmsummary_merge_override_basic(t->resources_allocated, limits);
-			debug(D_VINE, "%s (%s) busy on group '%s'", w->hostname, w->addrport, t->command_line);
-		} else {
-			rmsummary_delete(limits);
-		}
-	}
-	return result;
-}
-
 static void count_worker_resources(struct vine_manager *q, struct vine_worker_info *w)
 {
 	w->resources->cores.inuse = 0;
@@ -2865,6 +2830,7 @@ assignment and the new task state.
 
 static vine_result_code_t commit_task_to_worker(struct vine_manager *q, struct vine_worker_info *w, struct vine_task *t)
 {
+<<<<<<< HEAD
 	vine_result_code_t result = VINE_SUCCESS;
 
 	/* Kill unused libraries on this worker to reclaim resources. */
@@ -2894,31 +2860,52 @@ static vine_result_code_t commit_task_to_worker(struct vine_manager *q, struct v
 
 	t->time_when_commit_start = timestamp_get();
 	vine_result_code_t result;
+=======
+	vine_result_code_t result = 0;
+>>>>>>> 5ab8af678 (add worker code)
 	struct list *l = 0;
-	if (t->group_id) {
-		l = hash_table_lookup(q->task_group_table, t->group_id);
-	}
-	if (l && list_size(l) > 1) {
-		result = start_group_task(q, w, t, l);
-	} else {
+	l = hash_table_lookup(q->task_group_table, t->group_id);
+	int counter = 0;
+	do {	
+		/* Kill empty libraries to reclaim resources. Match the assumption of
+		 * @vine.schedule.c:check_worker_have_enough_resources() */
+		kill_empty_libraries_on_worker(q, w, t);
+		t->hostname = xxstrdup(w->hostname);
+		t->addrport = xxstrdup(w->addrport);
+
+		t->time_when_commit_start = timestamp_get();
 		result = start_one_task(q, w, t);
-	}
-	t->time_when_commit_end = timestamp_get();
+		t->time_when_commit_end = timestamp_get();
+
+		itable_insert(w->current_tasks, t->task_id, t);
+		t->worker = w;
+
+		/* Increment the function count if this is a function task.
+		 * If the manager fails to send this function task to the worker however,
+		 * then the count will be decremented properly in @handle_failure() below. */
+		if (t->needs_library) {
+			t->library_task = find_library_on_worker_for_task(w, t->needs_library);
+			t->library_task->function_slots_inuse++;
+			vine_txn_log_write_library_update(q, w, t->task_id, VINE_LIBRARY_SENT);
+		}
 
-	itable_insert(w->current_tasks, t->task_id, t);
-	t->worker = w;
+		change_task_state(q, t, VINE_TASK_RUNNING);
 
-	change_task_state(q, t, VINE_TASK_RUNNING);
+		t->try_count += 1;
+		q->stats->tasks_dispatched += 1;
 
-	t->try_count += 1;
-	q->stats->tasks_dispatched += 1;
+		count_worker_resources(q, w);
 
-	count_worker_resources(q, w);
+		if (result != VINE_SUCCESS) {
+			debug(D_VINE, "Failed to send task %d to worker %s (%s).", t->task_id, w->hostname, w->addrport);
+			handle_failure(q, w, t, result);
+		}
 
-	if (result != VINE_SUCCESS) {
-		debug(D_VINE, "Failed to send task %d to worker %s (%s).", t->task_id, w->hostname, w->addrport);
-		handle_failure(q, w, t, result);
-	}
+		counter++;
+	
+	} while((t = list_next_item(l)));
+
+	debug(D_VINE, "Sent batch of %d tasks to worker %s", counter, w->hostname);
 
 	return result;
 }

taskvine/src/manager/vine_manager_put.c

@@ -560,6 +560,10 @@ vine_result_code_t vine_manager_put_task(
 		}
 	}
 
+	if(t->group_id) {
+		vine_manager_send(q, w, "groupid %s\n", t->group_id);
+	}
+
 	// vine_manager_send returns the number of bytes sent, or a number less than
 	// zero to indicate errors. We are lazy here, we only check the last
 	// message we sent to the worker (other messages may have failed above).

taskvine/src/manager/vine_task_groups.c

@@ -19,7 +19,7 @@ static int vine_task_groups_create_group(struct vine_manager *q, struct vine_tas
 
 	t->group_id = id;
 
-	struct vine_task *tc = vine_task_copy(t);
+	struct vine_task *tc = vine_task_clone(t);
 
 	list_push_head(l, tc);
 	hash_table_insert(q->task_group_table, id, l);
@@ -33,13 +33,18 @@ static int vine_task_groups_add_to_group(struct vine_manager *q, struct vine_tas
 	char *id;
 	HASH_TABLE_ITERATE(q->task_group_table, id, l)
 	{
-		struct vine_file *f;
-		LIST_ITERATE(l, f)
+		struct vine_task *lt;
+		LIST_ITERATE(l, lt)
 		{
-			if (f == m->file) {
-				struct vine_task *tc = vine_task_copy(t);
-				list_push_tail(l, tc);
-				return 1;
+			struct vine_mount *lm;
+			LIST_ITERATE(lt->output_mounts, lm)
+			{
+				if (m->file == lm->file) {
+					t->group_id = lt->group_id; 
+					struct vine_task *tc = vine_task_clone(t);
+					list_push_tail(l, tc);
+					return 1;
+				}
 			}
 		}
 	}
@@ -78,8 +83,10 @@ int vine_task_groups_assign_task(struct vine_manager *q, struct vine_task *t)
 	// could also be inputs_present && outputs_present
 	if (inputs_present) {
 		vine_task_groups_add_to_group(q, t, input_mount);
+		debug(D_VINE, "Assigned task to group %s", t->group_id);
 	} else if (outputs_present) {
 		vine_task_groups_create_group(q, t, output_mount);
+		debug(D_VINE, "Create task with group %s", t->group_id);
 	}
 
 	return inputs_present || outputs_present;

taskvine/src/worker/vine_sandbox.c

@@ -43,7 +43,7 @@ vine_cache_status_t vine_sandbox_ensure(struct vine_process *p, struct vine_cach
 	LIST_ITERATE(p->task->input_mounts, m)
 	{
 		vine_cache_status_t cache_status = vine_cache_ensure(cache, m->file->cached_name);
-
+		
 		switch (cache_status) {
 		case VINE_CACHE_STATUS_PENDING:
 		case VINE_CACHE_STATUS_PROCESSING:
@@ -54,6 +54,11 @@ vine_cache_status_t vine_sandbox_ensure(struct vine_process *p, struct vine_cach
 			break;
 		case VINE_CACHE_STATUS_UNKNOWN:
 		case VINE_CACHE_STATUS_FAILED:
+			if(p->task->group_id)
+			{
+				processing++;
+				break;
+			}
 			return VINE_CACHE_STATUS_FAILED;
 		}
 	}

taskvine/src/worker/vine_worker.c

@@ -834,6 +834,7 @@ static struct vine_task *do_task_body(struct link *manager, int task_id, time_t
 	char taskname_encoded[VINE_LINE_MAX];
 	char library_name[VINE_LINE_MAX];
 	char category[VINE_LINE_MAX];
+	char groupid[VINE_LINE_MAX];
 	int flags, length;
 	int64_t n;
 
@@ -879,6 +880,8 @@ static struct vine_task *do_task_body(struct link *manager, int task_id, time_t
 			vine_task_set_disk(task, n);
 		} else if (sscanf(line, "gpus %" PRId64, &n)) {
 			vine_task_set_gpus(task, n);
+		} else if (sscanf(line, "groupid %s", groupid)) {
+			task->group_id = xxstrdup(groupid);	
 		} else if (sscanf(line, "wall_time %" PRIu64, &nt)) {
 			vine_task_set_time_max(task, nt);
 		} else if (sscanf(line, "end_time %" PRIu64, &nt)) {