starpu-max/src/core/simgrid.c

/* StarPU --- Runtime system for heterogeneous multicore architectures.
 *
 * Copyright (C) 2012-2017  Université de Bordeaux
 * Copyright (C) 2016  	    Inria
 * Copyright (C) 2016, 2017  	    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.
 */

#include <starpu.h>
#include <datawizard/memory_nodes.h>
#include <common/config.h>
#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif
#include <core/perfmodel/perfmodel.h>
#include <core/workers.h>
#include <core/simgrid.h>
#if defined(HAVE_SG_LINK_NAME) && (SIMGRID_VERSION_MAJOR >= 4 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR >= 13))
#include <simgrid/simdag.h>
#endif

#ifdef STARPU_SIMGRID
#include <sys/resource.h>
#include <simgrid/simix.h>

#pragma weak starpu_main
extern int starpu_main(int argc, char *argv[]);
#pragma weak smpi_main
extern int smpi_main(int (*realmain) (int argc, char *argv[]), int argc, char *argv[]);
#pragma weak _starpu_mpi_simgrid_init
extern int _starpu_mpi_simgrid_init(int argc, char *argv[]);

static int simgrid_started;

static int runners_running;
starpu_pthread_queue_t _starpu_simgrid_transfer_queue[STARPU_MAXNODES];
static struct transfer_runner {
	struct transfer *first_transfer, *last_transfer;
	msg_sem_t sem;
	msg_process_t runner;
} transfer_runner[STARPU_MAXNODES][STARPU_MAXNODES];
static int transfer_execute(int argc STARPU_ATTRIBUTE_UNUSED, char *argv[] STARPU_ATTRIBUTE_UNUSED);

starpu_pthread_queue_t _starpu_simgrid_task_queue[STARPU_NMAXWORKERS];
static struct worker_runner {
	struct task *first_task, *last_task;
	msg_sem_t sem;
	msg_process_t runner;
} worker_runner[STARPU_NMAXWORKERS];
static int task_execute(int argc STARPU_ATTRIBUTE_UNUSED, char *argv[] STARPU_ATTRIBUTE_UNUSED);

/* In case the MPI application didn't use smpicc to build the file containing
 * main(), try to cope by calling starpu_main */
int _starpu_smpi_simulated_main_(int argc, char *argv[])
{
	if (!starpu_main)
	{
		_STARPU_ERROR("In simgrid mode, the file containing the main() function of this application needs to be compiled with starpu.h or starpu_simgrid_wrap.h included, to properly rename it into starpu_main\n");
	}

	return starpu_main(argc, argv);
}
int smpi_simulated_main_(int argc, char *argv[]) __attribute__((weak, alias("_starpu_smpi_simulated_main_")));

#ifdef HAVE_MSG_ENVIRONMENT_GET_ROUTING_ROOT
#ifdef HAVE_MSG_GET_AS_BY_NAME
msg_as_t _starpu_simgrid_get_as_by_name(const char *name)
{
	return MSG_get_as_by_name(name);
}
#else /* HAVE_MSG_GET_AS_BY_NAME */
static msg_as_t __starpu_simgrid_get_as_by_name(msg_as_t root, const char *name)
{
	xbt_dict_t dict;
	xbt_dict_cursor_t cursor;
	const char *key;
	msg_as_t as, ret;
	dict = MSG_environment_as_get_routing_sons(root);
	xbt_dict_foreach(dict, cursor, key, as)
	{
		if (!strcmp(MSG_environment_as_get_name(as), name))
			return as;
		ret = __starpu_simgrid_get_as_by_name(as, name);
		if (ret)
			return ret;
	}
	return NULL;
}

msg_as_t _starpu_simgrid_get_as_by_name(const char *name)
{
	return __starpu_simgrid_get_as_by_name(MSG_environment_get_routing_root(), name);
}
#endif /* HAVE_MSG_GET_AS_BY_NAME */
#endif /* HAVE_MSG_ENVIRONMENT_GET_ROUTING_ROOT */

int _starpu_simgrid_get_nbhosts(const char *prefix)
{
	int ret;
	xbt_dynar_t hosts;
	unsigned i, nb;
	unsigned len = strlen(prefix);
#ifdef HAVE_MSG_ENVIRONMENT_GET_ROUTING_ROOT
	char new_prefix[32];

	if (_starpu_simgrid_running_smpi())
	{
		char name[32];
		STARPU_ASSERT(starpu_mpi_world_rank);
		snprintf(name, sizeof(name), STARPU_MPI_AS_PREFIX"%u", starpu_mpi_world_rank());
		hosts = MSG_environment_as_get_hosts(_starpu_simgrid_get_as_by_name(name));
		snprintf(new_prefix, sizeof(new_prefix), "%s-%s", name, prefix);
		prefix = new_prefix;
		len = strlen(prefix);
	}
	else
#endif /* HAVE_MSG_ENVIRONMENT_GET_ROUTING_ROOT */
		hosts = MSG_hosts_as_dynar();
	nb = xbt_dynar_length(hosts);

	ret = 0;
	for (i = 0; i < nb; i++)
	{
		const char *name;
		name = MSG_host_get_name(xbt_dynar_get_as(hosts, i, msg_host_t));
		if (!strncmp(name, prefix, len))
			ret++;
	}
	xbt_dynar_free(&hosts);
	return ret;
}

unsigned long long _starpu_simgrid_get_memsize(const char *prefix, unsigned devid)
{
	char name[32];
	msg_host_t host;
	const char *memsize;

	snprintf(name, sizeof(name), "%s%u", prefix, devid);

	host = _starpu_simgrid_get_host_by_name(name);
	if (!host)
		return 0;

	if (!MSG_host_get_properties(host))
		return 0;

	memsize = MSG_host_get_property_value(host, "memsize");
	if (!memsize)
		return 0;

	return atoll(memsize);
}

msg_host_t _starpu_simgrid_get_host_by_name(const char *name)
{
	if (_starpu_simgrid_running_smpi())
	{
		char mpiname[32];
		STARPU_ASSERT(starpu_mpi_world_rank);
		snprintf(mpiname, sizeof(mpiname), STARPU_MPI_AS_PREFIX"%d-%s", starpu_mpi_world_rank(), name);
		return MSG_get_host_by_name(mpiname);
	}
	else
		return MSG_get_host_by_name(name);
}

msg_host_t _starpu_simgrid_get_host_by_worker(struct _starpu_worker *worker)
{
	char *prefix;
	char name[16];
	msg_host_t host;
	switch (worker->arch)
	{
		case STARPU_CPU_WORKER:
			prefix = "CPU";
			break;
		case STARPU_CUDA_WORKER:
			prefix = "CUDA";
			break;
		case STARPU_OPENCL_WORKER:
			prefix = "OpenCL";
			break;
		default:
			STARPU_ASSERT(0);
	}
	snprintf(name, sizeof(name), "%s%d", prefix, worker->devid);
	host =  _starpu_simgrid_get_host_by_name(name);
	STARPU_ASSERT_MSG(host, "Could not find host %s!", name);
	return host;
}

static void start_simgrid(int *argc, char **argv)
{
	char path[256];

	simgrid_started = 1;

	if (!starpu_main && !(smpi_main && smpi_simulated_main_ != _starpu_smpi_simulated_main_))
	{
		_STARPU_ERROR("In simgrid mode, the file containing the main() function of this application needs to be compiled with starpu.h or starpu_simgrid_wrap.h included, to properly rename it into starpu_main\n");
	}

	MSG_init(argc, argv);
#if SIMGRID_VERSION_MAJOR < 3 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR < 9)
	/* Versions earlier than 3.9 didn't support our communication tasks */
	MSG_config("workstation/model", "ptask_L07");
#endif
	/* Simgrid uses tiny stacks by default.  This comes unexpected to our users.  */
	unsigned stack_size = 8192;
	struct rlimit rlim;
	if (getrlimit(RLIMIT_STACK, &rlim) == 0 && rlim.rlim_cur != 0 && rlim.rlim_cur != RLIM_INFINITY)
		stack_size = rlim.rlim_cur / 1024;

#if SIMGRID_VERSION_MAJOR < 3 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR < 13)
	extern xbt_cfg_t _sg_cfg_set;
	xbt_cfg_set_int(_sg_cfg_set, "contexts/stack_size", stack_size);
#else
	xbt_cfg_set_int("contexts/stack-size", stack_size);
#endif

	/* Load XML platform */
#if SIMGRID_VERSION_MAJOR < 3 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR < 13)
	_starpu_simgrid_get_platform_path(3, path, sizeof(path));
#else
	_starpu_simgrid_get_platform_path(4, path, sizeof(path));
#endif
	MSG_create_environment(path);
}

struct main_args
{
	int argc;
	char **argv;
};
static int main_ret;

int do_starpu_main(int argc, char *argv[])
{
	/* FIXME: Ugly work-around for bug in simgrid: the MPI context is not properly set at MSG process startup */
	MSG_process_sleep(0.000001);

	main_ret = starpu_main(argc, argv);
	return main_ret;
}

#undef main
#pragma weak main
int main(int argc, char **argv)
{
	if (_starpu_simgrid_running_smpi())
	{
		/* Oops, we are running SMPI, let it start Simgrid, and we'll
		 * take back hand in _starpu_simgrid_init from starpu_init() */
		return smpi_main(_starpu_mpi_simgrid_init, argc, argv);
	}

	/* Managed to catch application's main, initialize simgrid first */
	start_simgrid(&argc, argv);

	/* Create a simgrid process for main */
	char **argv_cpy;
	_STARPU_MALLOC(argv_cpy, argc * sizeof(char*));
	int i;
	for (i = 0; i < argc; i++)
		argv_cpy[i] = strdup(argv[i]);
	MSG_process_create_with_arguments("main", &do_starpu_main, calloc(MAX_TSD+1, sizeof(void*)), MSG_get_host_by_name("MAIN"), argc, argv_cpy);

	/* And run maestro in main thread */
	MSG_main();
	return main_ret;
}

#ifdef HAVE_MSG_PROCESS_ATTACH
static void maestro(void *data STARPU_ATTRIBUTE_UNUSED)
{
	MSG_main();
}
#endif

void _starpu_simgrid_init_early(int *argc STARPU_ATTRIBUTE_UNUSED, char ***argv STARPU_ATTRIBUTE_UNUSED)
{
#ifdef HAVE_MSG_PROCESS_ATTACH
	if (!simgrid_started && !(smpi_main && smpi_simulated_main_ != _starpu_smpi_simulated_main_))
	{
		_STARPU_DISP("Warning: In simgrid mode, the file containing the main() function of this application should to be compiled with starpu.h or starpu_simgrid_wrap.h included, to properly rename it into starpu_main to avoid having to use --cfg=contexts/factory:thread which reduces performance\n");
#if SIMGRID_VERSION_MAJOR > 3 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR >= 14)
		xbt_cfg_set_string("contexts/factory", "thread");
#endif
		/* We didn't catch application's main. */
		/* Start maestro as a separate thread */
		SIMIX_set_maestro(maestro, NULL);
		/* Initialize simgrid */
		start_simgrid(argc, *argv);
		/* And attach the main thread to the main simgrid process */
		MSG_process_attach("main", calloc(MAX_TSD, sizeof(void*)), MSG_get_host_by_name("MAIN"), NULL);
		simgrid_started = 2;
	}
#endif

	if (!simgrid_started && !starpu_main && !(smpi_main && smpi_simulated_main_ != _starpu_smpi_simulated_main_))
	{
		_STARPU_ERROR("In simgrid mode, the file containing the main() function of this application needs to be compiled with starpu.h or starpu_simgrid_wrap.h included, to properly rename it into starpu_main\n");
	}
	if (_starpu_simgrid_running_smpi())
	{
#ifndef STARPU_STATIC_ONLY
		_STARPU_ERROR("Simgrid currently does not support privatization for dynamically-linked libraries in SMPI. Please reconfigure and build StarPU with --disable-shared");
#endif
		MSG_process_set_data(MSG_process_self(), calloc(MAX_TSD, sizeof(void*)));
	}
	unsigned i;
	for (i = 0; i < STARPU_MAXNODES; i++)
		starpu_pthread_queue_init(&_starpu_simgrid_transfer_queue[i]);
	for (i = 0; i < STARPU_NMAXWORKERS; i++)
		starpu_pthread_queue_init(&_starpu_simgrid_task_queue[i]);
}

void _starpu_simgrid_init(void)
{
	unsigned i;
	runners_running = 1;
	for (i = 0; i < starpu_worker_get_count(); i++)
	{
		char s[32];
		snprintf(s, sizeof(s), "worker %u runner", i);
		void **tsd = calloc(MAX_TSD+1, sizeof(void*));
		worker_runner[i].sem = MSG_sem_init(0);
		tsd[0] = (void*)(uintptr_t) i;
		worker_runner[i].runner = MSG_process_create_with_arguments(s, task_execute, tsd, _starpu_simgrid_get_host_by_worker(_starpu_get_worker_struct(i)), 0, NULL);
	}
}

void _starpu_simgrid_deinit(void)
{
	unsigned i, j;
	runners_running = 0;
	for (i = 0; i < STARPU_MAXNODES; i++)
	{
		for (j = 0; j < STARPU_MAXNODES; j++)
		{
			struct transfer_runner *t = &transfer_runner[i][j];
			if (t->runner)
			{
				MSG_sem_release(t->sem);
#if SIMGRID_VERSION_MAJOR > 3 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR >= 14)
				MSG_process_join(t->runner, 1000000);
#else
				MSG_process_sleep(1);
#endif
				STARPU_ASSERT(t->first_transfer == NULL);
				STARPU_ASSERT(t->last_transfer == NULL);
				MSG_sem_destroy(t->sem);
			}
		}
		/* FIXME: queue not empty at this point, needs proper unregistration */
		/* starpu_pthread_queue_destroy(&_starpu_simgrid_transfer_queue[i]); */
	}
	for (i = 0; i < starpu_worker_get_count(); i++)
	{
		struct worker_runner *w = &worker_runner[i];
		MSG_sem_release(w->sem);
#if SIMGRID_VERSION_MAJOR > 3 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR >= 14)
		MSG_process_join(w->runner, 1000000);
#else
		MSG_process_sleep(1);
#endif
		STARPU_ASSERT(w->first_task == NULL);
		STARPU_ASSERT(w->last_task == NULL);
		MSG_sem_destroy(w->sem);
		starpu_pthread_queue_destroy(&_starpu_simgrid_task_queue[i]);
	}
#ifdef HAVE_MSG_PROCESS_ATTACH
	if (simgrid_started == 2)
	{
		/* Started with MSG_process_attach, now detach */
		MSG_process_detach();
		simgrid_started = 0;
	}
#endif
}

/*
 * Tasks
 */

struct task
{
	msg_task_t task;

	/* communication termination signalization */
	unsigned *finished;

	/* Next task on this worker */
	struct task *next;
};

/* Actually execute the task.  */
static int task_execute(int argc STARPU_ATTRIBUTE_UNUSED, char *argv[] STARPU_ATTRIBUTE_UNUSED)
{
	/* FIXME: Ugly work-around for bug in simgrid: the MPI context is not properly set at MSG process startup */
	MSG_process_sleep(0.000001);

	unsigned workerid = (uintptr_t) starpu_pthread_getspecific(0);
	struct worker_runner *w = &worker_runner[workerid];

	_STARPU_DEBUG("worker runner %u started\n", workerid);
	while (1) {
		struct task *task;

		MSG_sem_acquire(w->sem);
		if (!runners_running)
			break;

		task = w->first_task;
		w->first_task = task->next;
		if (w->last_task == task)
			w->last_task = NULL;

		_STARPU_DEBUG("task %p started\n", task);
		MSG_task_execute(task->task);
		MSG_task_destroy(task->task);
		_STARPU_DEBUG("task %p finished\n", task);

		*task->finished = 1;
		/* The worker which started this task may be sleeping out of tasks, wake it  */
		starpu_wake_worker(workerid);

		free(task);
	}
	_STARPU_DEBUG("worker %u stopped\n", workerid);
	return 0;
}

/* Wait for completion of all asynchronous tasks for this worker */
void _starpu_simgrid_wait_tasks(int workerid)
{
	struct task *task = worker_runner[workerid].last_task;
	if (!task)
		return;

	unsigned *finished = task->finished;
	starpu_pthread_wait_t wait;
	starpu_pthread_wait_init(&wait);
	starpu_pthread_queue_register(&wait, &_starpu_simgrid_task_queue[workerid]);

	while(1)
	{
		starpu_pthread_wait_reset(&wait);
		if (*finished)
			break;
		starpu_pthread_wait_wait(&wait);
	}
	starpu_pthread_queue_unregister(&wait, &_starpu_simgrid_task_queue[workerid]);
	starpu_pthread_wait_destroy(&wait);
}

/* Task execution submitted by StarPU */
void _starpu_simgrid_submit_job(int workerid, struct _starpu_job *j, struct starpu_perfmodel_arch* perf_arch, double length, unsigned *finished)
{
	struct starpu_task *starpu_task = j->task;
	msg_task_t simgrid_task;

	if (j->internal)
		/* This is not useful to include in simulation (and probably
		 * doesn't have a perfmodel anyway) */
		return;
	
	if (isnan(length))
	{
		length = starpu_task_expected_length(starpu_task, perf_arch, j->nimpl);
		STARPU_ASSERT_MSG(!_STARPU_IS_ZERO(length) && !isnan(length),
				"Codelet %s does not have a perfmodel, or is not calibrated enough, please re-run in non-simgrid mode until it is calibrated",
			_starpu_job_get_model_name(j));
	}

	simgrid_task = MSG_task_create(_starpu_job_get_task_name(j),
#ifdef HAVE_MSG_HOST_GET_SPEED
			length/1000000.0*MSG_host_get_speed(MSG_host_self()),
#else
			length/1000000.0*MSG_get_host_speed(MSG_host_self()),
#endif
			0, NULL);

	if (finished == NULL)
	{
		/* Synchronous execution */
		/* First wait for previous tasks */
		_starpu_simgrid_wait_tasks(workerid);
		MSG_task_execute(simgrid_task);
		MSG_task_destroy(simgrid_task);
	}
	else
	{
		/* Asynchronous execution */
		struct task *task;
		struct worker_runner *w = &worker_runner[workerid];
		_STARPU_MALLOC(task, sizeof(*task));
		task->task = simgrid_task;
		task->finished = finished;
		*finished = 0;
		task->next = NULL;
		/* Sleep 10µs for the GPU task queueing */
		if (_starpu_simgrid_queue_malloc_cost())
			MSG_process_sleep(0.000010);
		if (w->last_task)
		{
			/* Already running a task, queue */
			w->last_task->next = task;
			w->last_task = task;
		}
		else
		{
			STARPU_ASSERT(!w->first_task);
			w->first_task = task;
			w->last_task = task;
		}
		MSG_sem_release(w->sem);
	}
}

/*
 * Transfers
 */

/* Note: simgrid is not parallel, so there is no need to hold locks for management of transfers.  */
LIST_TYPE(transfer,
	msg_task_t task;
	int src_node;
	int dst_node;
	int run_node;

	/* communication termination signalization */
	unsigned *finished;

	/* transfers which wait for this transfer */
	struct transfer **wake;
	unsigned nwake;

	/* Number of transfers that this transfer waits for */
	unsigned nwait;

	/* Next transfer on this stream */
	struct transfer *next;
)

struct transfer_list pending;

/* Tell for two transfers whether they should be handled in sequence */
static int transfers_are_sequential(struct transfer *new_transfer, struct transfer *old_transfer)
{
	int new_is_cuda STARPU_ATTRIBUTE_UNUSED, old_is_cuda STARPU_ATTRIBUTE_UNUSED;
	int new_is_opencl STARPU_ATTRIBUTE_UNUSED, old_is_opencl STARPU_ATTRIBUTE_UNUSED;
	int new_is_gpu_gpu, old_is_gpu_gpu;

	new_is_cuda  = starpu_node_get_kind(new_transfer->src_node) == STARPU_CUDA_RAM;
	new_is_cuda |= starpu_node_get_kind(new_transfer->dst_node) == STARPU_CUDA_RAM;
	old_is_cuda  = starpu_node_get_kind(old_transfer->src_node) == STARPU_CUDA_RAM;
	old_is_cuda |= starpu_node_get_kind(old_transfer->dst_node) == STARPU_CUDA_RAM;

	new_is_opencl  = starpu_node_get_kind(new_transfer->src_node) == STARPU_OPENCL_RAM;
	new_is_opencl |= starpu_node_get_kind(new_transfer->dst_node) == STARPU_OPENCL_RAM;
	old_is_opencl  = starpu_node_get_kind(old_transfer->src_node) == STARPU_OPENCL_RAM;
	old_is_opencl |= starpu_node_get_kind(old_transfer->dst_node) == STARPU_OPENCL_RAM;

	new_is_gpu_gpu = new_transfer->src_node && new_transfer->dst_node;
	old_is_gpu_gpu = old_transfer->src_node && old_transfer->dst_node;

	/* We ignore cuda-opencl transfers, they can not happen */
	STARPU_ASSERT(!((new_is_cuda && old_is_opencl) || (old_is_cuda && new_is_opencl)));

	/* The following constraints have been observed with CUDA alone */

	/* Same source/destination, sequential */
	if (new_transfer->src_node == old_transfer->src_node && new_transfer->dst_node == old_transfer->dst_node)
		return 1;

	/* Crossed GPU-GPU, sequential */
	if (new_is_gpu_gpu
			&& new_transfer->src_node == old_transfer->dst_node
			&& old_transfer->src_node == new_transfer->dst_node)
		return 1;

	/* GPU-GPU transfers are sequential with any RAM->GPU transfer */
	if (new_is_gpu_gpu
			&& old_transfer->dst_node == new_transfer->src_node
			&& old_transfer->dst_node == new_transfer->dst_node)
		return 1;
	if (old_is_gpu_gpu
			&& new_transfer->dst_node == old_transfer->src_node
			&& new_transfer->dst_node == old_transfer->dst_node)
		return 1;

	/* StarPU's constraint on CUDA transfers is using one stream per
	 * source/destination pair, which is already handled above */

	return 0;
}

static void transfer_queue(struct transfer *transfer)
{
	unsigned src = transfer->src_node;
	unsigned dst = transfer->dst_node;
	struct transfer_runner *t = &transfer_runner[src][dst];

	if (!t->runner)
	{
		/* No runner yet, start it */
		static starpu_pthread_mutex_t mutex; /* process_create may yield */
		STARPU_PTHREAD_MUTEX_LOCK(&mutex);
		if (!t->runner)
		{
			char s[64];
			snprintf(s, sizeof(s), "transfer %u-%u runner", src, dst);
			void **tsd = calloc(MAX_TSD+1, sizeof(void*));
			tsd[0] = (void*)(uintptr_t)((src<<16) + dst);
			t->runner = MSG_process_create_with_arguments(s, transfer_execute, tsd, _starpu_simgrid_get_memnode_host(src), 0, NULL);
			t->sem = MSG_sem_init(0);
		}
		STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
	}

	if (t->last_transfer)
	{
		/* Already running a transfer, queue */
		t->last_transfer->next = transfer;
		t->last_transfer = transfer;
	}
	else
	{
		STARPU_ASSERT(!t->first_transfer);
		t->first_transfer = transfer;
		t->last_transfer = transfer;
	}
	MSG_sem_release(t->sem);
}

/* Actually execute the transfer, and then start transfers waiting for this one.  */
static int transfer_execute(int argc STARPU_ATTRIBUTE_UNUSED, char *argv[] STARPU_ATTRIBUTE_UNUSED)
{
	/* FIXME: Ugly work-around for bug in simgrid: the MPI context is not properly set at MSG process startup */
	MSG_process_sleep(0.000001);

	unsigned src_dst = (uintptr_t) starpu_pthread_getspecific(0);
	unsigned src = src_dst >> 16;
	unsigned dst = src_dst & 0xffff;
	struct transfer_runner *t = &transfer_runner[src][dst];

	_STARPU_DEBUG("transfer runner %u-%u started\n", src, dst);
	while (1) {
		struct transfer *transfer;

		MSG_sem_acquire(t->sem);
		if (!runners_running)
			break;
		transfer = t->first_transfer;
		t->first_transfer = transfer->next;
		if (t->last_transfer == transfer)
			t->last_transfer = NULL;

		_STARPU_DEBUG("transfer %p started\n", transfer);
		MSG_task_execute(transfer->task);
		MSG_task_destroy(transfer->task);
		_STARPU_DEBUG("transfer %p finished\n", transfer);

		*transfer->finished = 1;
		transfer_list_erase(&pending, transfer);

		/* The workers which started this request may be sleeping out of tasks, wake it  */
		_starpu_wake_all_blocked_workers_on_node(transfer->run_node);

		unsigned i;
		/* Wake transfers waiting for my termination */
		/* Note: due to possible preemption inside process_create, the array
		 * may grow while doing this */
		for (i = 0; i < transfer->nwake; i++)
		{
			struct transfer *wake = transfer->wake[i];
			STARPU_ASSERT(wake->nwait > 0);
			wake->nwait--;
			if (!wake->nwait)
			{
				_STARPU_DEBUG("triggering transfer %p\n", wake);
				transfer_queue(wake);
			}
		}
		free(transfer->wake);
		free(transfer);
	}

	return 0;
}

/* Look for sequentialization between this transfer and pending transfers, and submit this one */
static void transfer_submit(struct transfer *transfer)
{
	struct transfer *old;

	for (old  = transfer_list_begin(&pending);
	     old != transfer_list_end(&pending);
	     old  = transfer_list_next(old))
	{
		if (transfers_are_sequential(transfer, old))
		{
			_STARPU_DEBUG("transfer %p(%d->%d) waits for %p(%d->%d)\n",
					transfer, transfer->src_node, transfer->dst_node,
					old, old->src_node, old->dst_node);
			/* Make new wait for the old */
			transfer->nwait++;
			/* Make old wake the new */
			_STARPU_REALLOC(old->wake, (old->nwake + 1) * sizeof(old->wake));
			old->wake[old->nwake] = transfer;
			old->nwake++;
		}
	}

	transfer_list_push_front(&pending, transfer);

	if (!transfer->nwait)
	{
		_STARPU_DEBUG("transfer %p waits for nobody, starting\n", transfer);
		transfer_queue(transfer);
	}
}

int _starpu_simgrid_wait_transfer_event(union _starpu_async_channel_event *event)
{
	/* this is not associated to a request so it's synchronous */
	starpu_pthread_wait_t wait;
	starpu_pthread_wait_init(&wait);
	starpu_pthread_queue_register(&wait, event->queue);

	while(1)
	{
		starpu_pthread_wait_reset(&wait);
		if (event->finished)
			break;
		starpu_pthread_wait_wait(&wait);
	}
	starpu_pthread_queue_unregister(&wait, event->queue);
	starpu_pthread_wait_destroy(&wait);
	return 0;
}

int _starpu_simgrid_test_transfer_event(union _starpu_async_channel_event *event)
{
	return event->finished;
}

/* Data transfer issued by StarPU */
int _starpu_simgrid_transfer(size_t size, unsigned src_node, unsigned dst_node, struct _starpu_data_request *req)
{
	/* Simgrid does not like 0-bytes transfers */
	if (!size)
		return 0;

	msg_task_t task;
	msg_host_t *hosts;
	double *computation;
	double *communication;
	union _starpu_async_channel_event *event, myevent;

	_STARPU_CALLOC(hosts, 2, sizeof(*hosts));
	_STARPU_CALLOC(computation, 2, sizeof(*computation));
	_STARPU_CALLOC(communication, 4, sizeof(*communication));

	hosts[0] = _starpu_simgrid_memory_node_get_host(src_node);
	hosts[1] = _starpu_simgrid_memory_node_get_host(dst_node);
	STARPU_ASSERT(hosts[0] != hosts[1]);
	communication[1] = size;

	task = MSG_parallel_task_create("copy", 2, hosts, computation, communication, NULL);

	struct transfer *transfer = transfer_new();

	_STARPU_DEBUG("creating transfer %p for %lu bytes\n", transfer, (unsigned long) size);

	transfer->task = task;
	transfer->src_node = src_node;
	transfer->dst_node = dst_node;
	transfer->run_node = _starpu_memory_node_get_local_key();

	if (req)
		event = &req->async_channel.event;
	else
		event = &myevent;
	event->finished = 0;
	transfer->finished = &event->finished;
	event->queue = &_starpu_simgrid_transfer_queue[transfer->run_node];

	transfer->wake = NULL;
	transfer->nwake = 0;
	transfer->nwait = 0;
	transfer->next = NULL;

	if (req)
		_STARPU_TRACE_START_DRIVER_COPY_ASYNC(src_node, dst_node);

	/* Sleep 10µs for the GPU transfer queueing */
	if (_starpu_simgrid_queue_malloc_cost())
		MSG_process_sleep(0.000010);
	transfer_submit(transfer);
	/* Note: from here, transfer might be already freed */

	if (req)
	{
		_STARPU_TRACE_END_DRIVER_COPY_ASYNC(src_node, dst_node);
		_STARPU_TRACE_DATA_COPY(src_node, dst_node, size);
		return -EAGAIN;
	}
	else
	{
		/* this is not associated to a request so it's synchronous */
		_starpu_simgrid_wait_transfer_event(event);
		return 0;
	}
}

int
_starpu_simgrid_thread_start(int argc STARPU_ATTRIBUTE_UNUSED, char *argv[])
{
	void *(*f)(void*) = (void*) (uintptr_t) strtol(argv[0], NULL, 16);
	void *arg = (void*) (uintptr_t) strtol(argv[1], NULL, 16);

	/* FIXME: Ugly work-around for bug in simgrid: the MPI context is not properly set at MSG process startup */
	MSG_process_sleep(0.000001);

	/* _args is freed with process context */
	f(arg);
	return 0;
}

msg_host_t
_starpu_simgrid_get_memnode_host(unsigned node)
{
	const char *fmt;
	char name[16];

	switch (starpu_node_get_kind(node))
	{
		case STARPU_CPU_RAM:
			fmt = "RAM";
			break;
		case STARPU_CUDA_RAM:
			fmt = "CUDA%u";
			break;
		case STARPU_OPENCL_RAM:
			fmt = "OpenCL%u";
			break;
		default:
			STARPU_ABORT();
			break;
	}
	snprintf(name, sizeof(name), fmt, _starpu_memory_node_get_devid(node));

	return _starpu_simgrid_get_host_by_name(name);
}

void _starpu_simgrid_count_ngpus(void)
{
#if defined(HAVE_SG_LINK_NAME) && (SIMGRID_VERSION_MAJOR >= 4 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR >= 13))
	unsigned src, dst;
	msg_host_t ramhost = _starpu_simgrid_get_host_by_name("RAM");

	/* For each pair of memory nodes, get the route */
	for (src = 1; src < STARPU_MAXNODES; src++)
		for (dst = 1; dst < STARPU_MAXNODES; dst++)
		{
			int busid;
			msg_host_t srchost, dsthost;
			const SD_link_t *route;
			int i, routesize;
			int through;
			unsigned src2;
			unsigned ngpus;
			const char *name;

			if (dst == src)
				continue;
			busid = starpu_bus_get_id(src, dst);
			if (busid == -1)
				continue;

			srchost = _starpu_simgrid_get_memnode_host(src);
			dsthost = _starpu_simgrid_get_memnode_host(dst);
			routesize = SD_route_get_size(srchost, dsthost);
			route = SD_route_get_list(srchost, dsthost);

			/* If it goes through "Host", do not care, there is no
			 * direct transfer support */
			for (i = 0; i < routesize; i++)
				if (!strcmp(sg_link_name(route[i]), "Host"))
					break;
			if (i < routesize)
				continue;

			/* Get the PCI bridge between down and up links */
			through = -1;
			for (i = 0; i < routesize; i++)
			{
				name = sg_link_name(route[i]);
				size_t len = strlen(name);
				if (!strcmp(" through", name+len-8))
					through = i;
				else if (!strcmp(" up", name+len-3))
					break;
			}
			/* Didn't find it ?! */
			if (through == -1)
			{
				_STARPU_DEBUG("Didn't find through-link for %d->%d\n", src, dst);
				continue;
			}
			name = sg_link_name(route[through]);

			/*
			 * count how many direct routes go through it between
			 * GPUs and RAM
			 */
			ngpus = 0;
			for (src2 = 1; src2 < STARPU_MAXNODES; src2++)
			{
				if (starpu_bus_get_id(src2, STARPU_MAIN_RAM) == -1)
					continue;
				msg_host_t srchost2 = _starpu_simgrid_get_memnode_host(src2);
				int routesize2 = SD_route_get_size(srchost2, ramhost);
				const SD_link_t *route2 = SD_route_get_list(srchost2, ramhost);

				for (i = 0; i < routesize2; i++)
					if (!strcmp(name, sg_link_name(route2[i])))
					{
						/* This GPU goes through this PCI bridge to access RAM */
						ngpus++;
						break;
					}
			}
			_STARPU_DEBUG("%d->%d through %s, %u GPUs\n", src, dst, name, ngpus);
			starpu_bus_set_ngpus(busid, ngpus);
		}
#endif
}

typedef struct{
	void_f_pvoid_t code;
	void *userparam;
	void *father_data;
} thread_data_t;

static int _starpu_simgrid_xbt_thread_create_wrapper(int argc STARPU_ATTRIBUTE_UNUSED, char *argv[] STARPU_ATTRIBUTE_UNUSED)
{
	/* FIXME: Ugly work-around for bug in simgrid: the MPI context is not properly set at MSG process startup */
	MSG_process_sleep(0.000001);

#ifdef HAVE_SMX_ACTOR_T
	smx_actor_t
#else
	smx_process_t
#endif
	self = SIMIX_process_self();
	thread_data_t *t = SIMIX_process_self_get_data(self);
	simcall_process_set_data(self, t->father_data);
	t->code(t->userparam);
	simcall_process_set_data(self, NULL);
	free(t);

	return 0;
}

void _starpu_simgrid_xbt_thread_create(const char *name, void_f_pvoid_t code, void *param)
{
#ifdef HAVE_SMX_ACTOR_T
	smx_actor_t process STARPU_ATTRIBUTE_UNUSED;
#else
	smx_process_t process STARPU_ATTRIBUTE_UNUSED;
#endif
	thread_data_t *res;
	_STARPU_MALLOC(res, sizeof(thread_data_t));
	res->userparam = param;
	res->code = code;
	res->father_data = SIMIX_process_self_get_data(SIMIX_process_self());

#if SIMGRID_VERSION_MAJOR < 3 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR < 12)
	simcall_process_create(&process,
#else
	process = simcall_process_create(
#endif
	                         name,
	                         _starpu_simgrid_xbt_thread_create_wrapper, res,
#if SIMGRID_VERSION_MAJOR < 3 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR < 14)
	                         SIMIX_host_self_get_name(),
#else
	                         SIMIX_host_self(),
#endif
#if SIMGRID_VERSION_MAJOR < 3 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR < 15)
				 -1.0,
#endif
				 0, NULL,
	                         /*props */ NULL
#if SIMGRID_VERSION_MAJOR < 3 || (SIMGRID_VERSION_MAJOR == 3 && SIMGRID_VERSION_MINOR < 15)
				 , 0
#endif
				 );
}
#endif