starpu-max/src/util/starpu_clusters_create.c

/* StarPU --- Runtime system for heterogeneous multicore architectures.
 *
 * Copyright (C) 2015  Université de Bordeaux
 * Copyright (C) 2015  INRIA
 * Copyright (C) 2015  CNRS
 *
 * StarPU is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or (at
 * your option) any later version.
 *
 * StarPU is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 *
 * See the GNU Lesser General Public License in COPYING.LGPL for more details.
 */

/* This file creates an interface to manage clustering resources and make use
 * of parallel tasks. It entirely depends on the hwloc software. */

#include "starpu_clusters_create.h"

starpu_binding_function _starpu_cluster_type_get_func(starpu_cluster_types type)
{
		starpu_binding_function prologue_func;

		switch (type)
		{
		case OPENMP:
				prologue_func = &starpu_openmp_prologue;
				break;
		case INTEL_OPENMP_MKL:
				prologue_func = &starpu_intel_openmp_mkl_prologue;
				break;
#ifdef STARPU_MKL
		case GNU_OPENMP_MKL:
				prologue_func = &starpu_gnu_openmp_mkl_prologue;
				break;
#endif
		default:
				prologue_func = NULL;
		}

		return prologue_func;
}

void starpu_openmp_prologue(void * sched_ctx_id)
{
		int sched_ctx = *(int*)sched_ctx_id;
		int *cpuids = NULL;
		int ncpuids = 0;
		int workerid = starpu_worker_get_id();

		if (starpu_worker_get_type(workerid) == STARPU_CPU_WORKER)
		{
				starpu_sched_ctx_get_available_cpuids(sched_ctx, &cpuids, &ncpuids);
				omp_set_num_threads(ncpuids);
#pragma omp parallel
				{
						starpu_sched_ctx_bind_current_thread_to_cpuid(cpuids[omp_get_thread_num()]);
				}
				free(cpuids);
		}
		return;
}

#ifdef STARPU_MKL
void starpu_gnu_openmp_mkl_prologue(void * sched_ctx_id)
{
		int sched_ctx = *(int*)sched_ctx_id;
		int *cpuids = NULL;
		int ncpuids = 0;
		int workerid = starpu_worker_get_id();

		if (starpu_worker_get_type(workerid) == STARPU_CPU_WORKER)
		{
				starpu_sched_ctx_get_available_cpuids(sched_ctx, &cpuids, &ncpuids);
				omp_set_num_threads(ncpuids);
				mkl_set_num_threads(ncpuids);
				mkl_set_dynamic(0);
#pragma omp parallel
				{
						starpu_sched_ctx_bind_current_thread_to_cpuid(cpuids[omp_get_thread_num()]);
				}
				free(cpuids);
		}
		return;
}
#endif

/* Main interface function to create a cluster view of the machine.
 * Its job is to capture what the user wants and store it in a standard view. */
struct starpu_cluster_machine* starpu_cluster_machine(hwloc_obj_type_t cluster_level, ...)
{
		va_list varg_list;
		int arg_type;
		struct _starpu_cluster_parameters *params;
		struct starpu_cluster_machine* machine = malloc(sizeof(struct starpu_cluster_machine));
		machine->params = malloc(sizeof(struct _starpu_cluster_parameters));
		machine->groups = _starpu_cluster_group_list_new();

		_starpu_cluster_init_parameters(machine->params);
		params = machine->params;

		va_start(varg_list, cluster_level);
		while ((arg_type = va_arg(varg_list, int)) != 0)
		{
				if (arg_type == STARPU_CLUSTER_MIN_NB)
				{
						params->min_nb = va_arg(varg_list, int);
						if (params->min_nb <= 0)
								fprintf(stderr, "Caution min number of contexts shouldn't "
										"be negative or null\n");
				}
				else if (arg_type == STARPU_CLUSTER_MAX_NB)
				{
						params->max_nb = va_arg(varg_list, int);
						if (params->max_nb <= 0)
								fprintf(stderr, "Caution max number of contexts shouldn't "
										"be negative or null\n");
				}
				else if (arg_type == STARPU_CLUSTER_NB)
				{
						params->nb = va_arg(varg_list, int);
						if (params->nb <= 0)
								fprintf(stderr, "Caution number of contexts shouldn't "
										"be negative or null\n");
				}
				else if (arg_type == STARPU_CLUSTER_POLICY_NAME)
				{
						params->sched_policy_name = va_arg(varg_list, char*);
				}
				else if (arg_type == STARPU_CLUSTER_POLICY_STRUCT)
				{
						params->sched_policy_struct = va_arg(varg_list,
															 struct starpu_sched_policy*);
				}
				else if (arg_type == STARPU_CLUSTER_KEEP_HOMOGENEOUS)
				{
						params->keep_homogeneous = va_arg(varg_list, int); /* 0=off, other=on */
				}
				else if (arg_type == STARPU_CLUSTER_PREFERE_MIN)
				{
						params->prefere_min = va_arg(varg_list, int); /* 0=off, other=on */
				}
				else if (arg_type == STARPU_CLUSTER_CREATE_FUNC)
				{
						params->create_func = va_arg(varg_list, void (*)(void*));
				}
				else if (arg_type == STARPU_CLUSTER_CREATE_FUNC_ARG)
				{
						params->create_func_arg = va_arg(varg_list, void*);
				}
				else if (arg_type == STARPU_CLUSTER_TYPE)
				{
						params->type = va_arg(varg_list, starpu_cluster_types);
				}
				else if (arg_type == STARPU_CLUSTER_AWAKE_WORKERS)
				{
						params->awake_workers = va_arg(varg_list, unsigned);
				}
				else if (arg_type == STARPU_CLUSTER_PARTITION_ONE)
				{
						struct _starpu_cluster_group* group = _starpu_cluster_group_new();
						_starpu_cluster_group_init(group, machine);
						_starpu_cluster_group_list_push_back(machine->groups, group);
						params = group->params;
				}
				else if (arg_type == STARPU_CLUSTER_NEW)
				{
						struct _starpu_cluster* cluster = _starpu_cluster_new();
						struct _starpu_cluster_group* group = _starpu_cluster_group_list_back(machine->groups);
						if (group == NULL)
						{
								group = _starpu_cluster_group_new();
								_starpu_cluster_group_init(group, machine);
								_starpu_cluster_group_list_push_back(machine->groups, group);
						}
						_starpu_cluster_init(cluster, group);
						_starpu_cluster_list_push_back(group->clusters, cluster);
						params = cluster->params;
				}
				else if (arg_type == STARPU_CLUSTER_NCORES)
				{
						struct _starpu_cluster_group* group =
								_starpu_cluster_group_list_back(machine->groups);
						struct _starpu_cluster* cluster =
								_starpu_cluster_list_back(group->clusters);
						cluster->ncores = va_arg(varg_list, unsigned);

				}
				else
				{
						STARPU_ABORT_MSG("Unrecognized argument %d\n", arg_type);
				}
		}
		va_end(varg_list);

		switch(cluster_level)
		{
		case HWLOC_OBJ_MISC:
		case HWLOC_OBJ_BRIDGE:
		case HWLOC_OBJ_PCI_DEVICE:
		case HWLOC_OBJ_OS_DEVICE:
				STARPU_ABORT_MSG("Cluster aggregation isn't supported for level %s\n",
								 hwloc_obj_type_string(cluster_level));
				break;
		default: /* others can pass */
				break;
		}

		_starpu_cluster_machine(cluster_level, machine);
		return machine;
}

int starpu_uncluster_machine(struct starpu_cluster_machine* machine)
{
		struct _starpu_cluster_group *g, *tmp;
		struct _starpu_cluster_group_list* group_list = machine->groups;
		starpu_sched_ctx_delete(machine->id);
		g = _starpu_cluster_group_list_begin(group_list);
		while (g != _starpu_cluster_group_list_end(group_list))
		{
				tmp = g;
				g = _starpu_cluster_group_list_next(g);
				_starpu_cluster_group_remove(group_list, tmp);
		}
		hwloc_topology_destroy(machine->topology);
		free(machine->params);
		free(machine);
		starpu_sched_ctx_set_context(0);

		return 0;
}

void starpu_cluster_print(struct starpu_cluster_machine* clusters)
{
		int cnt, w;
		struct _starpu_cluster_group *group;
		struct _starpu_cluster *cluster;

		printf("Number of clusters created: %d\n", clusters->nclusters);
		cnt=0;
		for (group = _starpu_cluster_group_list_begin(clusters->groups);
			 group != _starpu_cluster_group_list_end(clusters->groups);
			 group = _starpu_cluster_group_list_next(group))
		{
				for (cluster = _starpu_cluster_list_begin(group->clusters);
					 cluster != _starpu_cluster_list_end(group->clusters);
					 cluster = _starpu_cluster_list_next(cluster))
				{
						printf("Cluster %d contains the following logical indexes:\n\t", cnt);
						for (w=0; w < cluster->ncores; w++)
								printf("%d ", cluster->cores[w]);
						printf("\n");
						cnt++;
				}
		}

		return;
}

void _starpu_cluster_create(struct _starpu_cluster* cluster)
{
		unsigned main_ctx_id = cluster->father->father->id;
		if (cluster->params->awake_workers)
				cluster->id = starpu_sched_ctx_create(cluster->workerids, cluster->ncores,
													  "clusters", STARPU_SCHED_CTX_NESTED,
													  main_ctx_id,
													  STARPU_SCHED_CTX_AWAKE_WORKERS,
													  0);
		else
				cluster->id = starpu_sched_ctx_create(cluster->workerids, cluster->ncores,
													  "clusters", STARPU_SCHED_CTX_NESTED,
													  main_ctx_id,
													  0);

		/* cluster priority can be the lowest, so let's enforce it */
		starpu_sched_ctx_set_priority(cluster->workerids, cluster->ncores, cluster->id, 0);
		return;
}

void _starpu_cluster_group_create(struct _starpu_cluster_group *group)
{
		struct _starpu_cluster* c;
		for (c = _starpu_cluster_list_begin(group->clusters) ;
			 c != _starpu_cluster_list_end(group->clusters) ;
			 c = _starpu_cluster_list_next(c))
		{
				_starpu_cluster_create(c);
				if (!c->params->awake_workers)
						_starpu_cluster_bind(c);
		}

		return;
}

int _starpu_cluster_bind(struct _starpu_cluster* cluster)
{
		starpu_binding_function func;
		void* func_arg;
		if (cluster->params->create_func)
		{
				func = cluster->params->create_func;
				func_arg = (void*) cluster->params->create_func_arg;
		}
		else
		{
				func = _starpu_cluster_type_get_func(cluster->params->type);
				func_arg = (void*) &cluster->id;
		}

		return starpu_task_insert(&_starpu_cluster_bind_cl,
								  STARPU_SCHED_CTX, cluster->id,
								  STARPU_POSSIBLY_PARALLEL, 1,
								  STARPU_PROLOGUE_CALLBACK_POP, func,
								  STARPU_PROLOGUE_CALLBACK_POP_ARG, func_arg,
								  0);
}

void _starpu_cluster_group_init(struct _starpu_cluster_group* group,
								struct starpu_cluster_machine* father)
{
		group->id = 0;
		group->nclusters = 0;
		group->clusters = _starpu_cluster_list_new();
		group->father = father;
		group->params = malloc(sizeof(struct _starpu_cluster_parameters));
		_starpu_cluster_copy_parameters(group->params,
										father->params);

		return;
}

void _starpu_cluster_init(struct _starpu_cluster* cluster,
						  struct _starpu_cluster_group* father)
{
		cluster->id = 0;
		cluster->cpuset = hwloc_bitmap_alloc();
		cluster->ncores = 0;
		cluster->cores = NULL;
		cluster->workerids = NULL;
		cluster->father = father;
		cluster->params = malloc(sizeof(struct _starpu_cluster_parameters));
		_starpu_cluster_copy_parameters(cluster->params,
										father->params);
}

int _starpu_cluster_remove(struct _starpu_cluster_list* cluster_list,
						   struct _starpu_cluster* cluster)
{
		if (cluster && cluster->id != STARPU_NMAX_SCHED_CTXS)
				starpu_sched_ctx_delete(cluster->id);
		else
				return -1;

		if (cluster->cores != NULL)
				free(cluster->cores);
		if (cluster->workerids != NULL)
				free(cluster->workerids);
		hwloc_bitmap_free(cluster->cpuset);
		free(cluster->params);
		_starpu_cluster_list_erase(cluster_list, cluster);
		_starpu_cluster_delete(cluster);

		return 0;
}

int _starpu_cluster_group_remove(struct _starpu_cluster_group_list* group_list,
								 struct _starpu_cluster_group* group)
{
		struct _starpu_cluster* tmp;
		struct _starpu_cluster_list* cluster_list = group->clusters;
		struct _starpu_cluster* c = _starpu_cluster_list_begin(cluster_list);
		while (c != _starpu_cluster_list_end(cluster_list))
		{
				tmp = c;
				c = _starpu_cluster_list_next(c);
				_starpu_cluster_remove(cluster_list, tmp);
		}
		free(group->params);
		_starpu_cluster_group_list_erase(group_list, group);
		_starpu_cluster_group_delete(group);

		return 0;
}

void _starpu_cluster_init_parameters(struct _starpu_cluster_parameters* params)
{
		params->min_nb = 0;
		params->max_nb = 0;
		params->nb = 0;
		params->sched_policy_name = NULL;
		params->sched_policy_struct = NULL;
		params->keep_homogeneous = 0;
		params->prefere_min = 0;
		params->create_func = NULL;
		params->create_func_arg = NULL;
		params->type = OPENMP;
		params->awake_workers = 0;

		return;
}

void _starpu_cluster_copy_parameters(struct _starpu_cluster_parameters* dst,
									 struct _starpu_cluster_parameters* src)
{
		dst->min_nb = src->min_nb;
		dst->max_nb = src->max_nb;
		dst->nb = src->nb;
		dst->sched_policy_name = src->sched_policy_name;
		dst->sched_policy_struct = src->sched_policy_struct;
		dst->keep_homogeneous = src->keep_homogeneous;
		dst->prefere_min = src->prefere_min;
		dst->create_func = src->create_func;
		dst->create_func_arg = src->create_func_arg;
		dst->type = src->type;
		dst->awake_workers = src->awake_workers;

		return;
}

/* Considering the resources and parameters, how many clusters should we take? */
int _starpu_cluster_analyze_parameters(struct _starpu_cluster_parameters* params, int npus)
{
		int nb_clusters = 1, j;
		if (params->nb)
		{
				nb_clusters = params->nb;
		}
		else if (params->min_nb && params->max_nb)
		{
				if (!params->keep_homogeneous)
				{
						if (params->prefere_min)
								nb_clusters = params->min_nb;
						else
								nb_clusters = params->max_nb;
				}
				else
				{
						int begin = params->prefere_min? params->min_nb:params->max_nb;
						int end = params->prefere_min? params->max_nb+1:params->min_nb-1;
						j=begin;
						int best = 0, second_best = 0, cpu_loss = INT_MAX;
						while (j != end)
						{
								if (npus%j == 0)
								{
										best = j;
										break;
								}
								if (npus%j < cpu_loss)
								{
										cpu_loss = npus%j;
										second_best = j;
								}
								j = params->prefere_min? j++:j--;
						}

						if (best)
								nb_clusters = best;
						else if (second_best)
								nb_clusters = second_best;
				}
		}

		return nb_clusters;
}

void _starpu_cluster_machine(hwloc_obj_type_t cluster_level,
							 struct starpu_cluster_machine* machine)
{
		struct _starpu_cluster_group* g;

		_starpu_cluster_topology(cluster_level, machine);

		if (machine->params->sched_policy_struct != NULL)
		{
				machine->id = starpu_sched_ctx_create(NULL, -1, "main sched ctx",
													  STARPU_SCHED_CTX_POLICY_STRUCT,
													  machine->params->sched_policy_struct,
													  0);
		}
		else if (machine->params->sched_policy_name != NULL)
		{
				machine->id = starpu_sched_ctx_create(NULL, -1, "main sched ctx",
													  STARPU_SCHED_CTX_POLICY_NAME,
													  machine->params->sched_policy_name,
													  0);
		}
		else
		{
				struct starpu_sched_policy* sched_policy;
				struct _starpu_sched_ctx* global_ctx =
						_starpu_get_sched_ctx_struct(STARPU_GLOBAL_SCHED_CTX);
				sched_policy = _starpu_get_sched_policy(global_ctx);
				machine->id = starpu_sched_ctx_create(NULL, -1, "main sched ctx",
													  STARPU_SCHED_CTX_POLICY_STRUCT,
													  sched_policy, 0);
		}


		for (g = _starpu_cluster_group_list_begin(machine->groups) ;
			 g != _starpu_cluster_group_list_end(machine->groups) ;
			 g = _starpu_cluster_group_list_next(g))
				_starpu_cluster_group_create(g);

		starpu_task_wait_for_all();
		starpu_sched_ctx_set_context(&machine->id);

		return;
}

void _starpu_cluster_topology(hwloc_obj_type_t cluster_level,
							  struct starpu_cluster_machine* machine)
{
		int w;
		hwloc_topology_t topology;
		hwloc_cpuset_t avail_cpus;
		char *buf;

		struct _starpu_machine_config* config = _starpu_get_machine_config();
		STARPU_ASSERT_MSG(config->topology.hwtopology != NULL, "STARPU_CLUSTER: You "
						  "need to call starpu_init() or make sure to activate hwloc.");
		hwloc_topology_dup(&topology, config->topology.hwtopology);

		avail_cpus = hwloc_bitmap_alloc();
		hwloc_bitmap_zero(avail_cpus);

		int nworkers = starpu_worker_get_count_by_type(STARPU_CPU_WORKER);
		int *workers = (int*) malloc(sizeof(int) * nworkers);
		starpu_worker_get_ids_by_type(STARPU_CPU_WORKER, workers, nworkers);

		for (w = 0; w < nworkers ; w++)
		{
				struct _starpu_worker *worker_str = _starpu_get_worker_struct(workers[w]);
				hwloc_bitmap_or(avail_cpus, avail_cpus, worker_str->hwloc_cpu_set);
		}
		hwloc_bitmap_list_asprintf(&buf, avail_cpus);

		hwloc_topology_restrict(topology, avail_cpus, 0);
		free(workers);

		/* Use new topology to fill in the cluster list */
		machine->topology = topology;
		_starpu_cluster_group(cluster_level, machine);

		hwloc_bitmap_free(avail_cpus);

		return;
}

void _starpu_cluster_group(hwloc_obj_type_t cluster_level,
						   struct starpu_cluster_machine* machine)
{
		unsigned nb_objects;
		int i;
		struct _starpu_cluster_group* group = NULL;

		if (machine->groups == NULL)
				machine->groups = _starpu_cluster_group_list_new();

		nb_objects = hwloc_get_nbobjs_by_type(machine->topology, cluster_level);
		if (nb_objects == 0)
				return;

		group = _starpu_cluster_group_list_begin(machine->groups);
		for (i = 0 ; i < nb_objects ; i++)
		{
				hwloc_obj_t cluster_obj = hwloc_get_obj_by_type(machine->topology,
																cluster_level, i);

				if (group == NULL)
				{
						group = _starpu_cluster_group_new();
						_starpu_cluster_group_init(group, machine);
						_starpu_cluster_group_list_push_back(machine->groups, group);
				}

				group->group_obj = cluster_obj;

				_starpu_cluster(group);
				machine->ngroups++;
				machine->nclusters += group->nclusters;
				group = _starpu_cluster_group_list_next(group);
		}

		return;
}

void _starpu_cluster(struct _starpu_cluster_group* group)
{
		int i, avail_pus, npus, npreset=0;
		struct _starpu_cluster* cluster;
		char* buf;
		npus = hwloc_get_nbobjs_inside_cpuset_by_type(group->father->topology,
													  group->group_obj->cpuset,
													  HWLOC_OBJ_PU);

		/* Preset clusters */
		avail_pus = npus;
		for (cluster=_starpu_cluster_list_begin(group->clusters);
			 cluster!=_starpu_cluster_list_end(group->clusters);
			 cluster=_starpu_cluster_list_next(cluster))
		{
				if (cluster->ncores > avail_pus)
						cluster->ncores = avail_pus;
				else if (avail_pus == 0)
						cluster->ncores = 0;

				if (cluster->ncores > 0)
				{
						cluster->cores = malloc(sizeof(int)*cluster->ncores);
						cluster->workerids = malloc(sizeof(int)*cluster->ncores);
						avail_pus -= cluster->ncores;
						npreset++;
				}
		}

		/* Automatic clusters */
		group->nclusters = _starpu_cluster_analyze_parameters(group->params, avail_pus);
		for (i=0 ; i<group->nclusters && avail_pus>0 ; i++)
		{
				if (cluster == NULL)
				{
						cluster = _starpu_cluster_new();
						_starpu_cluster_init(cluster, group);
						_starpu_cluster_list_push_back(group->clusters, cluster);
				}

				if (cluster->ncores != 0 && cluster->ncores > avail_pus)
				{
						cluster->ncores = avail_pus;
				}
				else
				{
						if (cluster->params->keep_homogeneous)
								cluster->ncores = avail_pus/(group->nclusters-i);
						else
								cluster->ncores = i==group->nclusters-1?
										avail_pus:
										avail_pus/(group->nclusters-i);
				}
				avail_pus -= cluster->ncores;
				cluster->cores = malloc(sizeof(int)*cluster->ncores);
				cluster->workerids = malloc(sizeof(int)*cluster->ncores);

				cluster = _starpu_cluster_list_next(cluster);
		}
		group->nclusters += npreset;

		cluster = _starpu_cluster_list_begin(group->clusters);
		int count = 0;
		static int starpu_cluster_warned = 0;
		hwloc_bitmap_list_asprintf(&buf, group->group_obj->cpuset);

		for (i=0 ; i<npus ; i++)
		{
				hwloc_obj_t pu = hwloc_get_obj_inside_cpuset_by_type(group->father->topology,
																	 group->group_obj->cpuset,
																	 HWLOC_OBJ_PU, i);

				/* If we have more than one worker on this resource, let's add them too --
				   even if it's bad (they'll all be boud on the same PU) */
				int size = 0, j;
				struct _starpu_worker* worker_str = _starpu_worker_list_front(pu->userdata);
				for (j = 0; j < _starpu_worker_list_size(pu->userdata) ; j++)
				{
						if (worker_str->arch == STARPU_CPU_WORKER)
								size++;
						worker_str = _starpu_worker_list_next(worker_str);
				}

				if (size > 1)
				{
						if (!starpu_cluster_warned)
						{
								fprintf(stderr, "STARPU CLUSTERS: Caution! It seems that you have"
								" multiple workers bound to the same PU. If you have"
								" multithreading on your cores it is greatly adviced"
								" to export STARPU_NTHREADS_PER_CORE=nb.");
								starpu_cluster_warned = 1;
						}
						cluster->ncores += size-1;
						cluster->cores = realloc(cluster->cores,
												 sizeof(int)*cluster->ncores);
						cluster->workerids = realloc(cluster->workerids,
													 sizeof(int)*cluster->ncores);
				}

				/* grab workerid list and return first cpu */
				worker_str = _starpu_worker_list_front(pu->userdata);
				if (worker_str)
						hwloc_bitmap_or(cluster->cpuset, cluster->cpuset,
										worker_str->hwloc_cpu_set);
				j = 0;
				while (worker_str)
				{
						if (worker_str->arch == STARPU_CPU_WORKER)
						{
								cluster->cores[count+j] = worker_str->bindid;
								cluster->workerids[count+j] = worker_str->workerid;
								j++;
						}
						worker_str = _starpu_worker_list_next(worker_str);
				}

				count+=size;
				if (cluster->ncores == count)
				{
						count = 0;
						cluster = _starpu_cluster_list_next(cluster);
				}
		}

		return;
}