Add hierarchical contexts

The hypervisor resizes contexts on level (only for feft policy)

include/starpu_sched_ctx.h

@@ -28,6 +28,7 @@ extern "C"
 #define STARPU_SCHED_CTX_POLICY_STRUCT		 (2<<16)
 #define STARPU_SCHED_CTX_POLICY_MIN_PRIO	 (3<<16)
 #define STARPU_SCHED_CTX_POLICY_MAX_PRIO	 (4<<16)
+#define STARPU_SCHED_CTX_HIERARCHY_LEVEL         (5<<16)
 
 unsigned starpu_sched_ctx_create(int *workerids_ctx, int nworkers_ctx, const char *sched_ctx_name, ...);
 
@@ -43,6 +44,10 @@ void starpu_sched_ctx_delete(unsigned sched_ctx_id);
 
 void starpu_sched_ctx_set_inheritor(unsigned sched_ctx_id, unsigned inheritor);
 
+unsigned starpu_sched_ctx_get_inheritor(unsigned sched_ctx_id);
+
+unsigned starpu_sched_ctx_get_hierarchy_level(unsigned sched_ctx_id);
+
 void starpu_sched_ctx_set_context(unsigned *sched_ctx_id);
 
 unsigned starpu_sched_ctx_get_context(void);

sc_hypervisor/examples/Makefile.am

@@ -21,7 +21,8 @@ AM_LDFLAGS = $(STARPU_OPENCL_LDFLAGS) $(STARPU_CUDA_LDFLAGS) $(STARPU_GLPK_LDFLA
 noinst_PROGRAMS =				\
 	app_driven_test/app_driven_test		\
 	lp_test/lp_test				\
-	lp_test/lp_resize_test
+	lp_test/lp_resize_test			\
+	hierarchical_ctxs/resize_hierarchical_ctxs
 
 if !NO_BLAS_LIB
 noinst_PROGRAMS +=				\

sc_hypervisor/include/sc_hypervisor.h

@@ -129,7 +129,13 @@ void sc_hypervisor_update_diff_total_flops(unsigned sched_ctx, double diff_total
 void sc_hypervisor_update_diff_elapsed_flops(unsigned sched_ctx, double diff_task_flops);
 
 /* updates the min and max workers needed by each context */
-	void sc_hypervisor_update_resize_interval(unsigned *sched_ctxs, int nsched_ctxs);
+void sc_hypervisor_update_resize_interval(unsigned *sched_ctxs, int nsched_ctxs);
+
+/* returns a list of contexts that are on the same level in the hierarchy of contexts */
+void sc_hypervisor_get_ctxs_on_level(unsigned **sched_ctxs, int *nsched_ctxs, unsigned hierarchy_level, unsigned father_sched_ctx_id);
+
+/* returns the number of levels of ctxs registered to the hyp */
+unsigned sc_hypervisor_get_nhierarchy_levels(void);
 #ifdef __cplusplus
 }
 #endif

sc_hypervisor/src/hypervisor_policies/feft_lp_policy.c

@@ -29,7 +29,7 @@ static void _try_resizing(unsigned *sched_ctxs, int nsched_ctxs, int *workers, i
 	starpu_trace_user_event(resize_no++);
 	int ns = sched_ctxs == NULL ? sc_hypervisor_get_nsched_ctxs() : nsched_ctxs;
 
-	if(ns <= 1) return;
+//	if(ns <= 1) return;
 
 	unsigned *curr_sched_ctxs = sched_ctxs == NULL ? sc_hypervisor_get_sched_ctxs() : sched_ctxs;
 	unsigned curr_nworkers = nworkers == -1 ? starpu_worker_get_count() : (unsigned)nworkers;
@@ -64,6 +64,32 @@ static void _try_resizing(unsigned *sched_ctxs, int nsched_ctxs, int *workers, i
 	}
 }
 
+static void _try_resizing_hierarchically(unsigned levels, unsigned current_level, unsigned *sched_ctxs, unsigned nsched_ctxs, int *pus, int npus)
+{
+	if(levels == 0)
+		return;
+
+	_try_resizing(sched_ctxs, nsched_ctxs, pus, npus);
+
+	int s;
+	for(s = 0; s < nsched_ctxs; s++)
+	{
+		unsigned *sched_ctxs_child;
+		int nsched_ctxs_child;
+		sc_hypervisor_get_ctxs_on_level(&sched_ctxs_child, &nsched_ctxs_child, current_level+1, sched_ctxs[s]);
+		if(nsched_ctxs_child > 0)
+		{
+			printf("ns %d on level %d \n", nsched_ctxs_child, current_level);
+			int *pus_father;
+			unsigned npus_father = 0;
+			npus_father = starpu_sched_ctx_get_workers_list(sched_ctxs[s], &pus_father);
+		
+			_try_resizing_hierarchically(levels-1, current_level+1, sched_ctxs_child, nsched_ctxs_child, pus_father, npus_father);
+
+			free(pus_father);
+		}
+	}
+}
 static void feft_lp_handle_poped_task(__attribute__((unused))unsigned sched_ctx, __attribute__((unused))int worker, 
 				      __attribute__((unused))struct starpu_task *task, __attribute__((unused))uint32_t footprint)
 {
@@ -76,7 +102,20 @@ static void feft_lp_handle_poped_task(__attribute__((unused))unsigned sched_ctx,
 		{
 			if(sc_hypervisor_check_speed_gap_btw_ctxs())
 			{
-				_try_resizing(NULL, -1, NULL, -1);
+				unsigned nhierarchy_levels = sc_hypervisor_get_nhierarchy_levels();
+				if(nhierarchy_levels > 1)
+				{
+					unsigned *sched_ctxs;
+					int nsched_ctxs;
+					sc_hypervisor_get_ctxs_on_level(&sched_ctxs, &nsched_ctxs, 0, STARPU_NMAX_SCHED_CTXS);
+					if(nsched_ctxs > 0)
+					{
+						_try_resizing_hierarchically(nhierarchy_levels, 0, sched_ctxs, nsched_ctxs, NULL, -1);
+						free(sched_ctxs);
+					}
+				}
+				else
+					_try_resizing(NULL, -1, NULL, -1);
 			}
 	
 			starpu_pthread_mutex_unlock(&act_hypervisor_mutex);
@@ -187,7 +226,30 @@ static void feft_lp_resize_ctxs(unsigned *sched_ctxs, int nsched_ctxs ,
 				 return;
 			 }
 		}
-		_try_resizing(sched_ctxs, nsched_ctxs, workers, nworkers);
+//		_try_resizing(sched_ctxs, nsched_ctxs, workers, nworkers);
+		unsigned nhierarchy_levels = sc_hypervisor_get_nhierarchy_levels();
+		if(nhierarchy_levels > 1)
+		{
+			if(nsched_ctxs == -1)
+			{
+				unsigned *sched_ctxs2;
+				int nsched_ctxs2;
+				sc_hypervisor_get_ctxs_on_level(&sched_ctxs2, &nsched_ctxs2, 0, STARPU_NMAX_SCHED_CTXS);
+				
+				if(nsched_ctxs2  > 0)
+				{
+					printf("ns = %d on 0 \n", nsched_ctxs2);
+					_try_resizing_hierarchically(nhierarchy_levels, 0, sched_ctxs2, nsched_ctxs2, workers, nworkers);
+					free(sched_ctxs2);
+				}
+			}
+			else
+				_try_resizing_hierarchically(nhierarchy_levels, 0, sched_ctxs, nsched_ctxs, workers, nworkers);
+		}
+		else
+//			_try_resizing(NULL, -1, NULL, -1);
+			_try_resizing(sched_ctxs, nsched_ctxs, workers, nworkers);
+
 		starpu_pthread_mutex_unlock(&act_hypervisor_mutex);
 	}
 }

sc_hypervisor/src/sc_hypervisor.c

@@ -802,7 +802,6 @@ void sc_hypervisor_update_resize_interval(unsigned *sched_ctxs, int nsched_ctxs)
 	unsigned sched_ctx;
 	int total_max_nworkers = 0;
 	int max_cpus = starpu_cpu_worker_get_count();
-	double max_workers_idle_time[nsched_ctxs];
 	unsigned configured = 0;
 	int i;
 	for(i = 0; i < nsched_ctxs; i++)
@@ -890,8 +889,8 @@ void sc_hypervisor_update_resize_interval(unsigned *sched_ctxs, int nsched_ctxs)
 		if(config->max_nworkers > max_cpus)
 			config->max_nworkers = max_cpus;
 		
-		printf("%d: ready tasks  %d idle for long %lf norm_idle_time %lf elapsed_time %lf norm_exec_time %lf nworker %d max %d \n", 
-		       sched_ctx, nready_tasks, max_workers_idle_time[i], norm_idle_time, elapsed_time, norm_exec_time, workers->nworkers, config->max_nworkers);
+		printf("%d: ready tasks  %d norm_idle_time %lf elapsed_time %lf norm_exec_time %lf nworker %d max %d \n", 
+		       sched_ctx, nready_tasks, norm_idle_time, elapsed_time, norm_exec_time, workers->nworkers, config->max_nworkers);
 
 
 		total_max_nworkers += config->max_nworkers;
@@ -1315,3 +1314,39 @@ void sc_hypervisor_update_diff_elapsed_flops(unsigned sched_ctx, double diff_ela
 //		starpu_pthread_mutex_unlock(&hypervisor.sched_ctx_w[sched_ctx].mutex);
 	}
 }
+
+void sc_hypervisor_get_ctxs_on_level(unsigned **sched_ctxs, int *nsched_ctxs, unsigned hierarchy_level, unsigned father_sched_ctx_id)
+{
+	unsigned s;
+	*nsched_ctxs = 0;
+	*sched_ctxs = (unsigned*)malloc(hypervisor.nsched_ctxs * sizeof(unsigned));
+	for(s = 0; s < hypervisor.nsched_ctxs; s++)
+	{
+		/* if father == STARPU_NMAX_SCHED_CTXS we take all the ctxs in this level */
+		if(starpu_sched_ctx_get_hierarchy_level(hypervisor.sched_ctxs[s]) == hierarchy_level && 
+		   (starpu_sched_ctx_get_inheritor(hypervisor.sched_ctxs[s]) == father_sched_ctx_id || father_sched_ctx_id == STARPU_NMAX_SCHED_CTXS))
+			(*sched_ctxs)[(*nsched_ctxs)++] = hypervisor.sched_ctxs[s];
+	}
+	if(*nsched_ctxs == 0)
+		free(*sched_ctxs);
+	return;
+}
+
+unsigned sc_hypervisor_get_nhierarchy_levels(void)
+{
+	unsigned nlevels = 0;
+	unsigned level = 0;
+	unsigned levels[STARPU_NMAX_SCHED_CTXS];
+	unsigned s, l;
+	for(s = 0; s < hypervisor.nsched_ctxs; s++)
+	{
+		level = starpu_sched_ctx_get_hierarchy_level(hypervisor.sched_ctxs[s]);
+		unsigned found = 0;
+		for(l = 0; l < nlevels; l++)
+			if(levels[l] == level)
+				found = 1;
+		if(!found)
+			levels[nlevels++] = level;
+	}
+	return nlevels;
+}

src/common/fxt.h

@@ -460,8 +460,7 @@ do {										\
 	FUT_DO_PROBE1(_STARPU_FUT_HYPERVISOR_BEGIN, _starpu_gettid());
 
 #define _STARPU_TRACE_HYPERVISOR_END() \
-	do {} while (0)
-//	FUT_DO_PROBE1(_STARPU_FUT_HYPERVISOR_END, _starpu_gettid());
+	FUT_DO_PROBE1(_STARPU_FUT_HYPERVISOR_END, _starpu_gettid());
 
 #ifdef STARPU_FXT_LOCK_TRACES 
 

src/core/sched_ctx.c

@@ -365,7 +365,7 @@ static void _get_workers(int min, int max, int *workers, int *nw, enum starpu_wo
 				{
 					int _npus = 0;
 					int _pus[STARPU_NMAXWORKERS];
-					_npus = starpu_get_workers_of_sched_ctx(config->sched_ctxs[s].id, _pus, arch);
+					_npus = _starpu_get_workers_of_sched_ctx(config->sched_ctxs[s].id, _pus, arch);
 					int ctx_min = arch == STARPU_CPU_WORKER ? config->sched_ctxs[s].min_ncpus : config->sched_ctxs[s].min_ngpus;
 					if(_npus > ctx_min)
 					{
@@ -411,7 +411,7 @@ static void _get_workers(int min, int max, int *workers, int *nw, enum starpu_wo
 					int _npus = 0;
 					int _pus[STARPU_NMAXWORKERS];
 
-					_npus = starpu_get_workers_of_sched_ctx(config->sched_ctxs[s].id, _pus, arch);
+					_npus = _starpu_get_workers_of_sched_ctx(config->sched_ctxs[s].id, _pus, arch);
 					if(needed_npus < (double)_npus)
 					{
 						double npus_to_rem = (double)_npus - needed_npus;
@@ -482,6 +482,7 @@ unsigned starpu_sched_ctx_create(int *workerids, int nworkers, const char *sched
 	int min_prio = 0;
 	int max_prio = 0;
 	struct starpu_sched_policy *sched_policy;
+	unsigned hierarchy_level = 0;
 
 	va_start(varg_list, sched_ctx_name);
 	while ((arg_type = va_arg(varg_list, int)) != 0)
@@ -506,6 +507,10 @@ unsigned starpu_sched_ctx_create(int *workerids, int nworkers, const char *sched
 			max_prio = va_arg(varg_list, int);
 			max_prio_set = 1;
 		}
+		else if (arg_type == STARPU_SCHED_CTX_HIERARCHY_LEVEL)
+		{
+			hierarchy_level = va_arg(varg_list, unsigned);
+		}
 		else
 		{
 			STARPU_ABORT_MSG("Unrecognized argument %d\n", arg_type);
@@ -516,6 +521,7 @@ unsigned starpu_sched_ctx_create(int *workerids, int nworkers, const char *sched
 
 	struct _starpu_sched_ctx *sched_ctx = NULL;
 	sched_ctx = _starpu_create_sched_ctx(sched_policy, workerids, nworkers, 0, sched_ctx_name, min_prio_set, min_prio, max_prio_set, max_prio);
+	sched_ctx->hierarchy_level = hierarchy_level;
 
 	_starpu_unlock_mutex_if_prev_locked();
 	_starpu_update_workers_with_ctx(sched_ctx->workers->workerids, sched_ctx->workers->nworkers, sched_ctx->id);
@@ -1077,7 +1083,7 @@ struct starpu_worker_collection* starpu_sched_ctx_get_worker_collection(unsigned
 	return sched_ctx->workers;
 }
 
-int starpu_get_workers_of_sched_ctx(unsigned sched_ctx_id, int *pus, enum starpu_worker_archtype arch)
+int _starpu_get_workers_of_sched_ctx(unsigned sched_ctx_id, int *pus, enum starpu_worker_archtype arch)
 {
 	struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
 
@@ -1093,7 +1099,7 @@ int starpu_get_workers_of_sched_ctx(unsigned sched_ctx_id, int *pus, enum starpu
 	{
 		worker = workers->get_next(workers, &it);
 		enum starpu_worker_archtype curr_arch = starpu_worker_get_type(worker);
-		if(curr_arch == arch)
+		if(curr_arch == arch || arch == STARPU_ANY_WORKER)
 			pus[npus++] = worker;
 	}
 
@@ -1220,12 +1226,29 @@ unsigned starpu_sched_ctx_overlapping_ctxs_on_worker(int workerid)
 
 void starpu_sched_ctx_set_inheritor(unsigned sched_ctx_id, unsigned inheritor)
 {
+	STARPU_ASSERT(sched_ctx_id < STARPU_NMAX_SCHED_CTXS);
 	STARPU_ASSERT(inheritor < STARPU_NMAX_SCHED_CTXS);
 	struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
 	sched_ctx->inheritor = inheritor;
 	return;
 }
 
+unsigned starpu_sched_ctx_get_inheritor(unsigned sched_ctx_id)
+{
+	STARPU_ASSERT(sched_ctx_id < STARPU_NMAX_SCHED_CTXS);
+	struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
+
+	return	sched_ctx->inheritor;
+}
+
+unsigned starpu_sched_ctx_get_hierarchy_level(unsigned sched_ctx_id)
+{
+	STARPU_ASSERT(sched_ctx_id < STARPU_NMAX_SCHED_CTXS);
+	struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
+
+	return	sched_ctx->hierarchy_level;
+}
+
 void starpu_sched_ctx_finished_submit(unsigned sched_ctx_id)
 {
 	struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);

src/core/sched_ctx.h

@@ -117,9 +117,12 @@ struct _starpu_sched_ctx
 	struct starpu_sched_ctx_performance_counters *perf_counters;
 #endif //STARPU_USE_SC_HYPERVISOR
 
+	/* callback called when the context finished executed its submitted tasks */
 	void (*close_callback)(unsigned sched_ctx_id, void* args);
-
 	void *close_args;
+	
+	/* value placing the contexts in their hierarchy */
+	unsigned hierarchy_level;
 };
 
 struct _starpu_machine_config;
@@ -161,7 +164,7 @@ starpu_pthread_mutex_t *_starpu_get_sched_mutex(struct _starpu_sched_ctx *sched_
 
 /* Get workers belonging to a certain context, it returns the number of workers
  take care: no mutex taken, the list of workers might not be updated */
-int starpu_get_workers_of_sched_ctx(unsigned sched_ctx_id, int *pus, enum starpu_worker_archtype arch);
+int _starpu_get_workers_of_sched_ctx(unsigned sched_ctx_id, int *pus, enum starpu_worker_archtype arch);
 
 /* Let the worker know it does not belong to the context and that
    it should stop poping from it */

src/debug/traces/starpu_fxt.c

@@ -698,13 +698,13 @@ static void handle_hyp_begin(struct fxt_ev_64 *ev, struct starpu_fxt_options *op
 
 static void handle_hyp_end(struct fxt_ev_64 *ev, struct starpu_fxt_options *options)
 {
-/* 	int worker; */
-/* 	worker = find_worker_id(ev->param[0]); */
-/* 	if (worker < 0) */
-/* 		return; */
+	int worker;
+	worker = find_worker_id(ev->param[0]);
+	if (worker < 0)
+		return;
 
-/* 	if (out_paje_file) */
-/* 		worker_set_state(get_event_time_stamp(ev, options), options->file_prefix, ev->param[0], "B"); */
+	if (out_paje_file)
+		worker_set_state(get_event_time_stamp(ev, options), options->file_prefix, ev->param[0], "B");
 }
 
 static void handle_worker_status(struct fxt_ev_64 *ev, struct starpu_fxt_options *options, const char *newstatus)

src/sched_policies/eager_central_policy.c

@@ -81,7 +81,6 @@ static int push_task_eager_policy(struct starpu_task *task)
 
 	starpu_push_task_end(task);
 
-
 	/*if there are no tasks block */
 	/* wake people waiting for a task */
 	unsigned worker = 0;