mpi: (manually) backport changes from 1.1 to support multiple MPI communicators

ChangeLog

@@ -156,6 +156,11 @@ New features:
   * New environment variable STARPU_PERF_MODEL_DIR which can be set to
     specify a directory where to store performance model files in.
     When unset, the files are stored in $STARPU_HOME/.starpu/sampling
+  * MPI:
+      - New function starpu_mpi_data_register_comm to register a data
+        with another communicator than MPI_COMM_WORLD
+      - New functions starpu_mpi_data_set_rank() and starpu_mpi_data_set_tag()
+        which call starpu_mpi_data_register_comm()
 
 Small features:
   * Add starpu_memory_wait_available() to wait for a given size to become

doc/doxygen/chapters/16mpi_support.doxy

@@ -1,7 +1,7 @@
 /*
  * This file is part of the StarPU Handbook.
  * Copyright (C) 2009--2011  Universit@'e de Bordeaux
- * Copyright (C) 2010, 2011, 2012, 2013, 2014  Centre National de la Recherche Scientifique
+ * Copyright (C) 2010, 2011, 2012, 2013, 2014, 2015  Centre National de la Recherche Scientifique
  * Copyright (C) 2011, 2012 Institut National de Recherche en Informatique et Automatique
  * See the file version.doxy for copying conditions.
  */
@@ -241,10 +241,8 @@ first needs to define a distribution function which specifies the
 locality of the data. Note that the data needs to be registered to MPI
 by calling starpu_mpi_data_register(). This function allows to set
 the distribution information and the MPI tag which should be used when
-communicating the data. The function starpu_mpi_data_register() should
-be prefered to starpu_data_set_rank() and starpu_data_set_tag() as
-it also allows to automatically clear the MPI communication cache
-when unregistering the data.
+communicating the data. It also allows to automatically clear the MPI
+communication cache when unregistering the data.
 
 \code{.c}
 /* Returns the MPI node number where data is */
@@ -435,7 +433,7 @@ migrate the data, and register the new location.
                 /* Migrate the data */
                 starpu_mpi_get_data_on_node_detached(MPI_COMM_WORLD, data_handles[x][y], mpi_rank, NULL, NULL);
                 /* And register the new rank of the matrix */
-                starpu_data_set_rank(data_handles[x][y], mpi_rank);
+                starpu_mpi_data_set_rank(data_handles[x][y], mpi_rank);
             }
         }
     }

doc/doxygen/chapters/api/mpi.doxy

@@ -217,29 +217,38 @@ function does nothing if the cache mechanism is disabled (see
 \anchor MPIInsertTask
 \ingroup API_MPI_Support
 
-\fn int starpu_data_set_tag(starpu_data_handle_t handle, int tag)
+\fn int starpu_mpi_data_register_comm(starpu_data_handle_t handle, int tag, int rank, MPI_Comm comm)
+\ingroup API_MPI_Support
+Register to MPI a StarPU data handle with the given tag, rank and MPI communicator.
+It also automatically clears the MPI communication cache when unregistering the data.
+
+\fn int starpu_mpi_data_register(starpu_data_handle_t handle, int tag, int rank)
 \ingroup API_MPI_Support
-Tell StarPU-MPI which MPI tag to use when exchanging the data.
+Register to MPI a StarPU data handle with the given tag, rank and the MPI communicator MPI_COMM_WORLD.
+It also automatically clears the MPI communication cache when unregistering the data.
 
-\fn int starpu_data_get_tag(starpu_data_handle_t handle)
+\fn int starpu_data_set_tag(starpu_data_handle_t handle, int tag)
 \ingroup API_MPI_Support
-Returns the MPI tag to be used when exchanging the data.
+Register to MPI a StarPU data handle with the given tag. No rank will be defined.
+It also automatically clears the MPI communication cache when unregistering the data.
 
-\fn void starpu_mpi_data_register(starpu_data_handle_t data_handle, int tag, int rank)
+\fn int starpu_data_set_rank_comm(starpu_data_handle_t handle, int rank, MPI_Comm comm)
 \ingroup API_MPI_Support
-Calling this function should be prefered to calling both
-starpu_data_set_rank() and starpu_data_set_tag() as it also allows to
-automatically clear the MPI communication cache when unregistering the data.
+Register to MPI a StarPU data handle with the given rank and given communicator. No tag will be defined.
+It also automatically clears the MPI communication cache when unregistering the data.
 
 \fn int starpu_data_set_rank(starpu_data_handle_t handle, int rank)
 \ingroup API_MPI_Support
-Tell StarPU-MPI which MPI node "owns" a given data, that is, the node
-which will always keep an up-to-date value, and will by default
-execute tasks which write to it.
+Register to MPI a StarPU data handle with the given rank and the MPI communicator MPI_COMM_WORLD. No tag will be defined.
+It also automatically clears the MPI communication cache when unregistering the data.
+
+\fn int starpu_mpi_data_get_rank(starpu_data_handle_t handle)
+\ingroup API_MPI_Support
+Return the rank of the given data.
 
-\fn int starpu_data_get_rank(starpu_data_handle_t handle)
+\fn int starpu_mpi_data_get_tag(starpu_data_handle_t handle)
 \ingroup API_MPI_Support
-Returns the last value set by starpu_data_set_rank().
+Return the tag of the given data.
 
 \def STARPU_EXECUTE_ON_NODE
 \ingroup API_MPI_Support

include/starpu_data.h

@@ -122,11 +122,6 @@ struct starpu_codelet;
 
 void starpu_data_set_reduction_methods(starpu_data_handle_t handle, struct starpu_codelet *redux_cl, struct starpu_codelet *init_cl);
 
-int starpu_data_set_rank(starpu_data_handle_t handle, int rank);
-int starpu_data_get_rank(starpu_data_handle_t handle);
-
-int starpu_data_set_tag(starpu_data_handle_t handle, int tag);
-int starpu_data_get_tag(starpu_data_handle_t handle);
 struct starpu_data_interface_ops* starpu_data_get_interface_ops(starpu_data_handle_t handle);
 
 unsigned starpu_data_test_if_allocated_on_node(starpu_data_handle_t handle, unsigned memory_node);

mpi/include/starpu_mpi.h

@@ -80,7 +80,19 @@ int starpu_mpi_world_rank(void);
 int starpu_mpi_get_communication_tag(void);
 void starpu_mpi_set_communication_tag(int tag);
 
-void starpu_mpi_data_register(starpu_data_handle_t data_handle, int tag, int rank);
+void starpu_mpi_data_register_comm(starpu_data_handle_t data_handle, int tag, int rank, MPI_Comm comm);
+#define starpu_mpi_data_register(data_handle, tag, rank) starpu_mpi_data_register_comm(data_handle, tag, rank, MPI_COMM_WORLD)
+
+void starpu_mpi_data_set_rank_comm(starpu_data_handle_t handle, int rank, MPI_Comm comm);
+#define starpu_mpi_data_set_rank(handle, rank) starpu_mpi_data_set_rank_comm(handle, rank, MPI_COMM_WORLD)
+void starpu_mpi_data_set_tag(starpu_data_handle_t handle, int tag);
+#define starpu_data_set_rank starpu_mpi_data_set_rank
+#define starpu_data_set_tag starpu_mpi_data_set_tag
+
+int starpu_mpi_data_get_rank(starpu_data_handle_t handle);
+int starpu_mpi_data_get_tag(starpu_data_handle_t handle);
+#define starpu_data_get_rank starpu_mpi_data_get_rank
+#define starpu_data_get_tag starpu_mpi_data_get_tag
 
 #define STARPU_MPI_NODE_SELECTION_CURRENT_POLICY -1
 #define STARPU_MPI_NODE_SELECTION_MOST_R_DATA    0

mpi/src/Makefile.am

@@ -42,7 +42,8 @@ noinst_HEADERS =					\
 	starpu_mpi_cache_stats.h			\
 	starpu_mpi_early_data.h				\
 	starpu_mpi_early_request.h			\
-	starpu_mpi_sync_data.h
+	starpu_mpi_sync_data.h				\
+	starpu_mpi_tag.h
 
 libstarpumpi_@STARPU_EFFECTIVE_VERSION@_la_SOURCES =	\
 	starpu_mpi.c					\
@@ -57,7 +58,8 @@ libstarpumpi_@STARPU_EFFECTIVE_VERSION@_la_SOURCES =	\
 	starpu_mpi_cache_stats.c			\
 	starpu_mpi_early_data.c				\
 	starpu_mpi_early_request.c			\
-	starpu_mpi_sync_data.c
+	starpu_mpi_sync_data.c				\
+	starpu_mpi_tag.c
 
 showcheck:
 	-cat /dev/null

mpi/src/starpu_mpi.c

@@ -19,6 +19,7 @@
 #include <starpu_mpi.h>
 #include <starpu_mpi_datatype.h>
 #include <starpu_mpi_private.h>
+#include <starpu_mpi_cache.h>
 #include <starpu_profiling.h>
 #include <starpu_mpi_stats.h>
 #include <starpu_mpi_cache.h>
@@ -26,6 +27,7 @@
 #include <starpu_mpi_early_data.h>
 #include <starpu_mpi_early_request.h>
 #include <starpu_mpi_select_node.h>
+#include <starpu_mpi_tag.h>
 #include <common/config.h>
 #include <common/thread.h>
 #include <datawizard/interfaces/data_interface.h>
@@ -324,27 +326,27 @@ static void _starpu_mpi_isend_data_func(struct _starpu_mpi_req *req)
 	if (req->sync == 0)
 	{
 		_STARPU_MPI_COMM_TO_DEBUG(req->count, req->datatype, req->srcdst, _STARPU_MPI_TAG_DATA, req->data_tag);
-		 req->ret = MPI_Isend(req->ptr, req->count, req->datatype, req->srcdst, _STARPU_MPI_TAG_DATA, req->comm, &req->request);
-		 STARPU_MPI_ASSERT_MSG(req->ret == MPI_SUCCESS, "MPI_Isend returning %s", _starpu_mpi_get_mpi_code(req->ret));
-	 }
-	 else
-	 {
+		req->ret = MPI_Isend(req->ptr, req->count, req->datatype, req->srcdst, _STARPU_MPI_TAG_DATA, req->comm, &req->request);
+		STARPU_MPI_ASSERT_MSG(req->ret == MPI_SUCCESS, "MPI_Isend returning %s", _starpu_mpi_get_mpi_code(req->ret));
+	}
+	else
+	{
 		_STARPU_MPI_COMM_TO_DEBUG(req->count, req->datatype, req->srcdst, _STARPU_MPI_TAG_SYNC_DATA, req->data_tag);
-		 req->ret = MPI_Issend(req->ptr, req->count, req->datatype, req->srcdst, _STARPU_MPI_TAG_SYNC_DATA, req->comm, &req->request);
-		 STARPU_MPI_ASSERT_MSG(req->ret == MPI_SUCCESS, "MPI_Issend returning %s", _starpu_mpi_get_mpi_code(req->ret));
-	 }
+		req->ret = MPI_Issend(req->ptr, req->count, req->datatype, req->srcdst, _STARPU_MPI_TAG_SYNC_DATA, req->comm, &req->request);
+		STARPU_MPI_ASSERT_MSG(req->ret == MPI_SUCCESS, "MPI_Issend returning %s", _starpu_mpi_get_mpi_code(req->ret));
+	}
 
-	 _STARPU_MPI_TRACE_ISEND_SUBMIT_END(req->srcdst, req->data_tag, 0);
+	_STARPU_MPI_TRACE_ISEND_SUBMIT_END(req->srcdst, req->data_tag, 0);
 
-	 /* somebody is perhaps waiting for the MPI request to be posted */
-	 STARPU_PTHREAD_MUTEX_LOCK(&req->req_mutex);
-	 req->submitted = 1;
-	 STARPU_PTHREAD_COND_BROADCAST(&req->req_cond);
-	 STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
+	/* somebody is perhaps waiting for the MPI request to be posted */
+	STARPU_PTHREAD_MUTEX_LOCK(&req->req_mutex);
+	req->submitted = 1;
+	STARPU_PTHREAD_COND_BROADCAST(&req->req_cond);
+	STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
 
-	 _starpu_mpi_handle_detached_request(req);
+	_starpu_mpi_handle_detached_request(req);
 
-	 _STARPU_MPI_LOG_OUT();
+	_STARPU_MPI_LOG_OUT();
 }
 
 static void _starpu_mpi_isend_size_func(struct _starpu_mpi_req *req)
@@ -1148,7 +1150,7 @@ static void _starpu_mpi_print_thread_level_support(int thread_level, char *msg)
 	}
 }
 
-static void _starpu_mpi_receive_early_data(struct _starpu_mpi_envelope *envelope, MPI_Status status)
+static void _starpu_mpi_receive_early_data(struct _starpu_mpi_envelope *envelope, MPI_Status status, MPI_Comm comm)
 {
 	_STARPU_MPI_DEBUG(20, "Request with tag %d and source %d not found, creating a early_handle to receive incoming data..\n", envelope->data_tag, status.MPI_SOURCE);
 	_STARPU_MPI_DEBUG(20, "Request sync %d\n", envelope->sync);
@@ -1157,7 +1159,7 @@ static void _starpu_mpi_receive_early_data(struct _starpu_mpi_envelope *envelope
 
 	starpu_data_handle_t data_handle = NULL;
 	STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
-	data_handle = _starpu_data_get_data_handle_from_tag(envelope->data_tag);
+	data_handle = _starpu_mpi_data_get_data_handle_from_tag(envelope->data_tag);
 	STARPU_PTHREAD_MUTEX_LOCK(&mutex);
 
 	if (data_handle && starpu_data_get_interface_id(data_handle) < STARPU_MAX_INTERFACE_ID)
@@ -1182,7 +1184,7 @@ static void _starpu_mpi_receive_early_data(struct _starpu_mpi_envelope *envelope
 	_STARPU_MPI_DEBUG(20, "Posting internal detached irecv on early_handle with tag %d from src %d ..\n", early_data_handle->data_tag, status.MPI_SOURCE);
 	STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
 	early_data_handle->req = _starpu_mpi_irecv_common(early_data_handle->handle, status.MPI_SOURCE,
-							  early_data_handle->data_tag, MPI_COMM_WORLD, 1, 0,
+							  early_data_handle->data_tag, comm, 1, 0,
 							  NULL, NULL, 1, 1, envelope->size);
 	STARPU_PTHREAD_MUTEX_LOCK(&mutex);
 
@@ -1248,6 +1250,7 @@ static void *_starpu_mpi_progress_thread_func(void *arg)
 	_starpu_mpi_comm_amounts_init(MPI_COMM_WORLD);
 	_starpu_mpi_cache_init(MPI_COMM_WORLD);
 	_starpu_mpi_select_node_init();
+	_starpu_mpi_tag_init();
 
 	_starpu_mpi_early_request_init(worldsize);
 	_starpu_mpi_early_data_init(worldsize);
@@ -1384,7 +1387,7 @@ static void *_starpu_mpi_progress_thread_func(void *arg)
 						}
 						else
 						{
-							_starpu_mpi_receive_early_data(envelope, status);
+							_starpu_mpi_receive_early_data(envelope, status, MPI_COMM_WORLD);
 						}
 					}
 					/* Case: a matching application request has been found for
@@ -1626,29 +1629,65 @@ int starpu_mpi_shutdown(void)
 	_starpu_mpi_comm_amounts_display(rank);
 	_starpu_mpi_comm_amounts_free();
 	_starpu_mpi_cache_free(world_size);
+	_starpu_mpi_tag_free();
 
 	return 0;
 }
 
 void _starpu_mpi_clear_cache(starpu_data_handle_t data_handle)
 {
-	starpu_mpi_cache_flush(MPI_COMM_WORLD, data_handle);
+	_starpu_mpi_data_release_tag(data_handle);
+	struct _starpu_mpi_data *mpi_data = data_handle->mpi_data;
+	_starpu_mpi_cache_flush(mpi_data->comm, data_handle);
+	free(data_handle->mpi_data);
 }
 
-void starpu_mpi_data_register(starpu_data_handle_t data_handle, int tag, int rank)
+void starpu_mpi_data_register_comm(starpu_data_handle_t data_handle, int tag, int rank, MPI_Comm comm)
 {
-#ifdef STARPU_DEVEL
-#warning see if the following code is really needed, it deadlocks some applications
-#if 0
-	int my;
-	starpu_mpi_comm_rank(MPI_COMM_WORLD, &my);
-	if (my != rank)
-		STARPU_MPI_ASSERT_MSG(data_handle->home_node == -1, "Data does not belong to node %d, it should be assigned a home node -1", my);
-#endif
-#endif
-	_starpu_data_set_rank(data_handle, rank);
-	_starpu_data_set_tag(data_handle, tag);
-	_starpu_data_set_unregister_hook(data_handle, _starpu_mpi_clear_cache);
+	struct _starpu_mpi_data *mpi_data;
+	if (data_handle->mpi_data)
+	{
+		mpi_data = data_handle->mpi_data;
+	}
+	else
+	{
+		mpi_data = malloc(sizeof(struct _starpu_mpi_data));
+		data_handle->mpi_data = mpi_data;
+		_starpu_mpi_data_register_tag(data_handle, tag);
+		_starpu_data_set_unregister_hook(data_handle, _starpu_mpi_clear_cache);
+	}
+
+	if (tag != -1)
+	{
+		mpi_data->tag = tag;
+	}
+	if (rank != -1)
+	{
+		mpi_data->rank = rank;
+		mpi_data->comm = comm;
+	}
+}
+
+void starpu_mpi_data_set_rank_comm(starpu_data_handle_t handle, int rank, MPI_Comm comm)
+{
+	starpu_mpi_data_register_comm(handle, -1, rank, comm);
+}
+
+void starpu_mpi_data_set_tag(starpu_data_handle_t handle, int tag)
+{
+	starpu_mpi_data_register_comm(handle, tag, -1, MPI_COMM_WORLD);
+}
+
+int starpu_mpi_data_get_rank(starpu_data_handle_t data)
+{
+	STARPU_ASSERT_MSG(data->mpi_data, "starpu_mpi_data_register MUST be called for data %p\n", data);
+	return ((struct _starpu_mpi_data *)(data->mpi_data))->rank;
+}
+
+int starpu_mpi_data_get_tag(starpu_data_handle_t data)
+{
+	STARPU_ASSERT_MSG(data->mpi_data, "starpu_mpi_data_register MUST be called for data %p\n", data);
+	return ((struct _starpu_mpi_data *)(data->mpi_data))->tag;
 }
 
 int starpu_mpi_comm_size(MPI_Comm comm, int *size)

mpi/src/starpu_mpi_cache.c

@@ -173,7 +173,7 @@ void _starpu_mpi_cache_sent_data_clear(MPI_Comm comm, starpu_data_handle_t data)
 
 void _starpu_mpi_cache_received_data_clear(starpu_data_handle_t data)
 {
-	int mpi_rank = starpu_data_get_rank(data);
+	int mpi_rank = starpu_mpi_data_get_rank(data);
 	struct _starpu_data_entry *already_received;
 
 	STARPU_PTHREAD_MUTEX_LOCK(&_cache_received_mutex[mpi_rank]);
@@ -209,7 +209,7 @@ void starpu_mpi_cache_flush_all_data(MPI_Comm comm)
 		STARPU_PTHREAD_MUTEX_LOCK(&_cache_sent_mutex[i]);
 		HASH_ITER(hh, _cache_sent_data[i], entry, tmp)
 		{
-			mpi_rank = starpu_data_get_rank(entry->data);
+			mpi_rank = starpu_mpi_data_get_rank(entry->data);
 			if (mpi_rank != my_rank && mpi_rank != -1)
 				starpu_data_invalidate_submit(entry->data);
 			HASH_DEL(_cache_sent_data[i], entry);
@@ -220,7 +220,7 @@ void starpu_mpi_cache_flush_all_data(MPI_Comm comm)
 		STARPU_PTHREAD_MUTEX_LOCK(&_cache_received_mutex[i]);
 		HASH_ITER(hh, _cache_received_data[i], entry, tmp)
 		{
-			mpi_rank = starpu_data_get_rank(entry->data);
+			mpi_rank = starpu_mpi_data_get_rank(entry->data);
 			if (mpi_rank != my_rank && mpi_rank != -1)
 				starpu_data_invalidate_submit(entry->data);
 			HASH_DEL(_cache_received_data[i], entry);
@@ -231,7 +231,7 @@ void starpu_mpi_cache_flush_all_data(MPI_Comm comm)
 	}
 }
 
-void starpu_mpi_cache_flush(MPI_Comm comm, starpu_data_handle_t data_handle)
+void _starpu_mpi_cache_flush(MPI_Comm comm, starpu_data_handle_t data_handle)
 {
 	struct _starpu_data_entry *avail;
 	int i, my_rank, nb_nodes;
@@ -241,7 +241,7 @@ void starpu_mpi_cache_flush(MPI_Comm comm, starpu_data_handle_t data_handle)
 
 	starpu_mpi_comm_size(comm, &nb_nodes);
 	starpu_mpi_comm_rank(comm, &my_rank);
-	mpi_rank = starpu_data_get_rank(data_handle);
+	mpi_rank = starpu_mpi_data_get_rank(data_handle);
 
 	for(i=0 ; i<nb_nodes ; i++)
 	{
@@ -266,7 +266,16 @@ void starpu_mpi_cache_flush(MPI_Comm comm, starpu_data_handle_t data_handle)
 		}
 		STARPU_PTHREAD_MUTEX_UNLOCK(&_cache_received_mutex[i]);
 	}
+}
+
+void starpu_mpi_cache_flush(MPI_Comm comm, starpu_data_handle_t data_handle)
+{
+	int my_rank, mpi_rank;
+
+	_starpu_mpi_cache_flush(comm, data_handle);
 
+	MPI_Comm_rank(comm, &my_rank);
+	mpi_rank = starpu_mpi_data_get_rank(data_handle);
 	if (mpi_rank != my_rank && mpi_rank != -1)
 		starpu_data_invalidate_submit(data_handle);
 }

mpi/src/starpu_mpi_cache.h

@@ -46,6 +46,8 @@ void _starpu_mpi_cache_received_data_clear(starpu_data_handle_t data);
 void *_starpu_mpi_cache_sent_data_set(starpu_data_handle_t data, int dest);
 void _starpu_mpi_cache_sent_data_clear(MPI_Comm comm, starpu_data_handle_t data);
 
+void _starpu_mpi_cache_flush(MPI_Comm comm, starpu_data_handle_t data_handle);
+
 #ifdef __cplusplus
 }
 #endif

mpi/src/starpu_mpi_collective.c

@@ -1,6 +1,6 @@
 /* StarPU --- Runtime system for heterogeneous multicore architectures.
  *
- * Copyright (C) 2011, 2012, 2013, 2014  Centre National de la Recherche Scientifique
+ * Copyright (C) 2011, 2012, 2013, 2014, 2015  Centre National de la Recherche Scientifique
  *
  * 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
@@ -63,8 +63,8 @@ int starpu_mpi_scatter_detached(starpu_data_handle_t *data_handles, int count, i
 		{
 			if (data_handles[x])
 			{
-				int owner = starpu_data_get_rank(data_handles[x]);
-				int data_tag = starpu_data_get_tag(data_handles[x]);
+				int owner = starpu_mpi_data_get_rank(data_handles[x]);
+				int data_tag = starpu_mpi_data_get_tag(data_handles[x]);
 				STARPU_ASSERT_MSG(data_tag >= 0, "Invalid tag for data handle");
 				if ((rank == root) && (owner != root))
 				{
@@ -82,8 +82,8 @@ int starpu_mpi_scatter_detached(starpu_data_handle_t *data_handles, int count, i
 	{
 		if (data_handles[x])
 		{
-			int owner = starpu_data_get_rank(data_handles[x]);
-			int data_tag = starpu_data_get_tag(data_handles[x]);
+			int owner = starpu_mpi_data_get_rank(data_handles[x]);
+			int data_tag = starpu_mpi_data_get_tag(data_handles[x]);
 			STARPU_ASSERT_MSG(data_tag >= 0, "Invalid tag for data handle");
 			if ((rank == root) && (owner != root))
 			{
@@ -125,8 +125,8 @@ int starpu_mpi_gather_detached(starpu_data_handle_t *data_handles, int count, in
 		{
 			if (data_handles[x])
 			{
-				int owner = starpu_data_get_rank(data_handles[x]);
-				int data_tag = starpu_data_get_tag(data_handles[x]);
+				int owner = starpu_mpi_data_get_rank(data_handles[x]);
+				int data_tag = starpu_mpi_data_get_tag(data_handles[x]);
 				STARPU_ASSERT_MSG(data_tag >= 0, "Invalid tag for data handle");
 				if ((rank == root) && (owner != root))
 				{
@@ -144,8 +144,8 @@ int starpu_mpi_gather_detached(starpu_data_handle_t *data_handles, int count, in
 	{
 		if (data_handles[x])
 		{
-			int owner = starpu_data_get_rank(data_handles[x]);
-			int data_tag = starpu_data_get_tag(data_handles[x]);
+			int owner = starpu_mpi_data_get_rank(data_handles[x]);
+			int data_tag = starpu_mpi_data_get_tag(data_handles[x]);
 			STARPU_ASSERT_MSG(data_tag >= 0, "Invalid tag for data handle");
 			if ((rank == root) && (owner != root))
 			{

mpi/src/starpu_mpi_private.h

@@ -208,6 +208,13 @@ LIST_TYPE(_starpu_mpi_req,
 	int sequential_consistency;
 );
 
+struct _starpu_mpi_data
+{
+	int tag;
+	int rank;
+	MPI_Comm comm;
+};
+
 #ifdef __cplusplus
 }
 #endif

mpi/src/starpu_mpi_tag.c

@@ -0,0 +1,115 @@
+/* StarPU --- Runtime system for heterogeneous multicore architectures.
+ *
+ * Copyright (C) 2011, 2012, 2013, 2014, 2015  Centre National de la Recherche Scientifique
+ * Copyright (C) 2011-2015  Université de Bordeaux
+ * Copyright (C) 2014 INRIA
+ *
+ * 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 <starpu_mpi.h>
+#include <starpu_mpi_private.h>
+#include <common/uthash.h>
+#include <common/starpu_spinlock.h>
+#include <datawizard/coherency.h>
+
+/* Entry in the `registered_tag_handles' hash table.  */
+struct handle_tag_entry
+{
+	UT_hash_handle hh;
+	int tag;
+	starpu_data_handle_t handle;
+};
+
+/* Hash table mapping host tags to data handles.  */
+static struct handle_tag_entry *registered_tag_handles;
+static struct _starpu_spinlock    registered_tag_handles_lock;
+
+void _starpu_mpi_tag_init(void)
+{
+	_starpu_spin_init(&registered_tag_handles_lock);
+}
+
+void _starpu_mpi_tag_free(void)
+{
+     	struct handle_tag_entry *tag_entry, *tag_tmp;
+
+	_starpu_spin_destroy(&registered_tag_handles_lock);
+
+	HASH_ITER(hh, registered_tag_handles, tag_entry, tag_tmp)
+	{
+		HASH_DEL(registered_tag_handles, tag_entry);
+		free(tag_entry);
+	}
+
+	registered_tag_handles = NULL;
+}
+
+starpu_data_handle_t _starpu_mpi_data_get_data_handle_from_tag(int tag)
+{
+	struct handle_tag_entry *ret;
+
+	_starpu_spin_lock(&registered_tag_handles_lock);
+	HASH_FIND_INT(registered_tag_handles, &tag, ret);
+	_starpu_spin_unlock(&registered_tag_handles_lock);
+
+	if (ret)
+	{
+		return ret->handle;
+	}
+	else
+	{
+		return NULL;
+	}
+}
+
+void _starpu_mpi_data_register_tag(starpu_data_handle_t handle, int tag)
+{
+	struct handle_tag_entry *entry;
+	entry = (struct handle_tag_entry *) malloc(sizeof(*entry));
+	STARPU_ASSERT(entry != NULL);
+
+	STARPU_ASSERT_MSG(!(_starpu_mpi_data_get_data_handle_from_tag(tag)),
+			  "There is already a data handle %p registered with the tag %d\n", _starpu_mpi_data_get_data_handle_from_tag(tag), tag);
+
+	_STARPU_MPI_DEBUG(42, "Adding handle %p with tag %d in hashtable\n", handle, tag);
+
+	entry->handle = handle;
+	entry->tag = tag;
+
+	_starpu_spin_lock(&registered_tag_handles_lock);
+	HASH_ADD_INT(registered_tag_handles, tag, entry);
+	_starpu_spin_unlock(&registered_tag_handles_lock);
+}
+
+int _starpu_mpi_data_release_tag(starpu_data_handle_t handle)
+{
+	struct handle_tag_entry *tag_entry;
+	int tag = starpu_mpi_data_get_tag(handle);
+
+	_STARPU_MPI_DEBUG(42, "Removing handle %p with tag %d from hashtable\n", handle, tag);
+
+	if (tag != -1)
+	{
+		_starpu_spin_lock(&registered_tag_handles_lock);
+		HASH_FIND_INT(registered_tag_handles, &(((struct _starpu_mpi_data *)(handle->mpi_data))->tag), tag_entry);
+		STARPU_ASSERT_MSG((tag_entry != NULL),"Data handle %p with tag %d isn't in the hashmap !",handle,tag);
+
+		HASH_DEL(registered_tag_handles, tag_entry);
+
+		_starpu_spin_unlock(&registered_tag_handles_lock);
+
+		free(tag_entry);
+	}
+	return 0;
+}

mpi/src/starpu_mpi_tag.h

@@ -0,0 +1,38 @@
+/* StarPU --- Runtime system for heterogeneous multicore architectures.
+ *
+ * Copyright (C) 2015  Centre National de la Recherche Scientifique
+ *
+ * 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.
+ */
+
+#ifndef __STARPU_MPI_TAG_H__
+#define __STARPU_MPI_TAG_H__
+
+#include <starpu.h>
+#include <stdlib.h>
+#include <mpi.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+void _starpu_mpi_tag_init(void);
+void _starpu_mpi_tag_free(void);
+void _starpu_mpi_data_register_tag(starpu_data_handle_t handle, int tag);
+int _starpu_mpi_data_release_tag(starpu_data_handle_t handle);
+starpu_data_handle_t _starpu_mpi_data_get_data_handle_from_tag(int tag);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // __STARPU_MPI_TAG_H__

mpi/src/starpu_mpi_task_insert.c

@@ -54,7 +54,7 @@ int _starpu_mpi_find_executee_node(starpu_data_handle_t data, enum starpu_data_a
 			return -EINVAL;
 		}
 
-		int mpi_rank = starpu_data_get_rank(data);
+		int mpi_rank = starpu_mpi_data_get_rank(data);
 		if (mpi_rank == -1)
 		{
 			_STARPU_ERROR("Data %p with mode STARPU_W needs to have a valid rank", data);
@@ -100,16 +100,16 @@ void _starpu_mpi_exchange_data_before_execution(starpu_data_handle_t data, enum
 {
 	if (data && mode & STARPU_R)
 	{
-		int mpi_rank = starpu_data_get_rank(data);
-		int data_tag = starpu_data_get_tag(data);
+		int mpi_rank = starpu_mpi_data_get_rank(data);
+		int data_tag = starpu_mpi_data_get_tag(data);
 		if (mpi_rank == -1)
 		{
-			fprintf(stderr,"StarPU needs to be told the MPI rank of this data, using starpu_data_set_rank\n");
+			fprintf(stderr,"StarPU needs to be told the MPI rank of this data, using starpu_mpi_data_register\n");
 			STARPU_ABORT();
 		}
 		if (data_tag == -1)
 		{
-			fprintf(stderr,"StarPU needs to be told the MPI tag of this data, using starpu_data_set_tag\n");
+			fprintf(stderr,"StarPU needs to be told the MPI tag of this data, using starpu_mpi_data_register\n");
 			STARPU_ABORT();
 		}
 
@@ -144,16 +144,16 @@ void _starpu_mpi_exchange_data_after_execution(starpu_data_handle_t data, enum s
 {
 	if (mode & STARPU_W)
 	{
-		int mpi_rank = starpu_data_get_rank(data);
-		int data_tag = starpu_data_get_tag(data);
+		int mpi_rank = starpu_mpi_data_get_rank(data);
+		int data_tag = starpu_mpi_data_get_tag(data);
 		if(mpi_rank == -1)
 		{
-			fprintf(stderr,"StarPU needs to be told the MPI rank of this data, using starpu_data_set_rank\n");
+			fprintf(stderr,"StarPU needs to be told the MPI rank of this data, using starpu_mpi_data_register\n");
 			STARPU_ABORT();
 		}
 		if(data_tag == -1)
 		{
-			fprintf(stderr,"StarPU needs to be told the MPI tag of this data, using starpu_data_set_tag\n");
+			fprintf(stderr,"StarPU needs to be told the MPI tag of this data, using starpu_mpi_data_register\n");
 			STARPU_ABORT();
 		}
 		if (mpi_rank == me)
@@ -189,7 +189,7 @@ void _starpu_mpi_clear_data_after_execution(starpu_data_handle_t data, enum star
 		/* We allocated a temporary buffer for the received data, now drop it */
 		if ((mode & STARPU_R) && do_execute)
 		{
-			int mpi_rank = starpu_data_get_rank(data);
+			int mpi_rank = starpu_mpi_data_get_rank(data);
 			if (mpi_rank != me && mpi_rank != -1)
 			{
 				starpu_data_invalidate_submit(data);
@@ -235,7 +235,7 @@ int _starpu_mpi_task_decode_v(struct starpu_codelet *codelet, int me, int nb_nod
 			starpu_data_handle_t data = va_arg(varg_list_copy, starpu_data_handle_t);
 			if (node_selected == 0)
 			{
-				*xrank = starpu_data_get_rank(data);
+				*xrank = starpu_mpi_data_get_rank(data);
 				STARPU_ASSERT_MSG(*xrank != -1, "Rank of the data must be set using starpu_mpi_data_register() or starpu_data_set_rank()");
 				_STARPU_MPI_DEBUG(100, "Executing on data node %d\n", *xrank);
 				STARPU_ASSERT_MSG(*xrank <= nb_nodes, "Node %d to execute codelet is not a valid node (%d)", *xrank, nb_nodes);
@@ -587,15 +587,15 @@ void starpu_mpi_get_data_on_node_detached(MPI_Comm comm, starpu_data_handle_t da
 {
 	int me, rank, tag;
 
-	rank = starpu_data_get_rank(data_handle);
-	tag = starpu_data_get_tag(data_handle);
+	rank = starpu_mpi_data_get_rank(data_handle);
+	tag = starpu_mpi_data_get_tag(data_handle);
 	if (rank == -1)
 	{
-		_STARPU_ERROR("StarPU needs to be told the MPI rank of this data, using starpu_mpi_data_register() or starpu_data_set_rank()\n");
+		_STARPU_ERROR("StarPU needs to be told the MPI rank of this data, using starpu_mpi_data_register() or starpu_mpi_data_register()\n");
 	}
 	if (tag == -1)
 	{
-		_STARPU_ERROR("StarPU needs to be told the MPI tag of this data, using starpu_mpi_data_register() or starpu_data_set_tag()\n");
+		_STARPU_ERROR("StarPU needs to be told the MPI tag of this data, using starpu_mpi_data_register() or starpu_mpi_data_register()\n");
 	}
 	starpu_mpi_comm_rank(comm, &me);
 
@@ -615,16 +615,16 @@ void starpu_mpi_get_data_on_node(MPI_Comm comm, starpu_data_handle_t data_handle
 {
 	int me, rank, tag;
 
-	rank = starpu_data_get_rank(data_handle);
-	tag = starpu_data_get_tag(data_handle);
+	rank = starpu_mpi_data_get_rank(data_handle);
+	tag = starpu_mpi_data_get_tag(data_handle);
 	if (rank == -1)
 	{
-		fprintf(stderr,"StarPU needs to be told the MPI rank of this data, using starpu_data_set_rank\n");
+		fprintf(stderr,"StarPU needs to be told the MPI rank of this data, using starpu_mpi_data_register\n");
 		STARPU_ABORT();
 	}
 	if (tag == -1)
 	{
-		fprintf(stderr,"StarPU needs to be told the MPI tag of this data, using starpu_data_set_tag\n");
+		fprintf(stderr,"StarPU needs to be told the MPI tag of this data, using starpu_mpi_data_register\n");
 		STARPU_ABORT();
 	}
 	starpu_mpi_comm_rank(comm, &me);
@@ -705,16 +705,16 @@ void starpu_mpi_redux_data(MPI_Comm comm, starpu_data_handle_t data_handle)
 {
 	int me, rank, tag, nb_nodes;
 
-	rank = starpu_data_get_rank(data_handle);
-	tag = starpu_data_get_tag(data_handle);
+	rank = starpu_mpi_data_get_rank(data_handle);
+	tag = starpu_mpi_data_get_tag(data_handle);
 	if (rank == -1)
 	{
-		fprintf(stderr,"StarPU needs to be told the MPI rank of this data, using starpu_data_set_rank\n");
+		fprintf(stderr,"StarPU needs to be told the MPI rank of this data, using starpu_mpi_data_register\n");
 		STARPU_ABORT();
 	}
 	if (tag == -1)
 	{
-		fprintf(stderr,"StarPU needs to be told the MPI tag of this data, using starpu_data_set_tag\n");
+		fprintf(stderr,"StarPU needs to be told the MPI tag of this data, using starpu_mpi_data_register\n");
 		STARPU_ABORT();
 	}
 

src/datawizard/coherency.h

@@ -1,7 +1,7 @@
 /* StarPU --- Runtime system for heterogeneous multicore architectures.
  *
  * Copyright (C) 2009-2015  Université de Bordeaux
- * Copyright (C) 2010, 2011, 2012, 2013, 2014  Centre National de la Recherche Scientifique
+ * Copyright (C) 2010, 2011, 2012, 2013, 2014, 2015  Centre National de la Recherche Scientifique
  * Copyright (C) 2014  Inria
  *
  * StarPU is free software; you can redistribute it and/or modify
@@ -223,8 +223,7 @@ struct _starpu_data_state
 #endif
 
         /* Used for MPI */
-        int rank;
-	int tag;
+	void *mpi_data;
 
 	_starpu_memory_stats_t memory_stats;
 

src/datawizard/filters.c

@@ -2,7 +2,7 @@
  *
  * Copyright (C) 2010-2014  Université de Bordeaux
  * Copyright (C) 2010  Mehdi Juhoor <mjuhoor@gmail.com>
- * Copyright (C) 2010, 2011, 2012, 2013  Centre National de la Recherche Scientifique
+ * Copyright (C) 2010, 2011, 2012, 2013, 2015  Centre National de la Recherche Scientifique
  * Copyright (C) 2012 INRIA
  *
  * StarPU is free software; you can redistribute it and/or modify
@@ -162,7 +162,7 @@ void starpu_data_partition(starpu_data_handle_t initial_handle, struct starpu_da
 		STARPU_ASSERT(child);
 
 		child->nchildren = 0;
-                child->rank = initial_handle->rank;
+                child->mpi_data = initial_handle->mpi_data;
 		child->root_handle = initial_handle->root_handle;
 		child->father_handle = initial_handle;
 		child->sibling_index = i;

src/datawizard/interfaces/data_interface.c

@@ -1,7 +1,7 @@
 /* StarPU --- Runtime system for heterogeneous multicore architectures.
  *
  * Copyright (C) 2009-2014  Université de Bordeaux
- * Copyright (C) 2010, 2011, 2012, 2013, 2014  Centre National de la Recherche Scientifique
+ * Copyright (C) 2010, 2011, 2012, 2013, 2014, 2015  Centre National de la Recherche Scientifique
  * Copyright (C) 2014  Inria
  *
  * StarPU is free software; you can redistribute it and/or modify
@@ -42,24 +42,11 @@ static struct handle_entry *registered_handles;
 static struct _starpu_spinlock    registered_handles_lock;
 static int _data_interface_number = STARPU_MAX_INTERFACE_ID;
 
-/* Entry in the `registered_tag_handles' hash table.  */
-struct handle_tag_entry
-{
-	UT_hash_handle hh;
-	int tag;
-	starpu_data_handle_t handle;
-};
-
-/* Hash table mapping host tags to data handles.  */
-static struct handle_tag_entry *registered_tag_handles;
-static struct _starpu_spinlock    registered_tag_handles_lock;
-
 static void _starpu_data_unregister(starpu_data_handle_t handle, unsigned coherent, unsigned nowait);
 
 void _starpu_data_interface_init(void)
 {
 	_starpu_spin_init(&registered_handles_lock);
-	_starpu_spin_init(&registered_tag_handles_lock);
 }
 
 void _starpu_data_interface_shutdown()
@@ -80,18 +67,6 @@ void _starpu_data_interface_shutdown()
 	}
 
 	registered_handles = NULL;
-
-	struct handle_tag_entry *tag_entry, *tag_tmp;
-
-	_starpu_spin_destroy(&registered_tag_handles_lock);
-
-	HASH_ITER(hh, registered_tag_handles, tag_entry, tag_tmp)
-	{
-		HASH_DEL(registered_tag_handles, tag_entry);
-		free(tag_entry);
-	}
-
-	registered_tag_handles = NULL;
 }
 
 #ifdef STARPU_OPENMP
@@ -275,8 +250,7 @@ static void _starpu_register_new_data(starpu_data_handle_t handle,
 	handle->father_handle = NULL;
 	handle->sibling_index = 0; /* could be anything for the root */
 	handle->depth = 1; /* the tree is just a node yet */
-        handle->rank = -1; /* invalid until set */
-	handle->tag = -1; /* invalid until set */
+        handle->mpi_data = NULL; /* invalid until set */
 
 	handle->is_not_important = 0;
 
@@ -405,6 +379,7 @@ int _starpu_data_handle_init(starpu_data_handle_t handle, struct starpu_data_int
 
 	handle->ops = interface_ops;
 	handle->mf_node = mf_node;
+	handle->mpi_data = NULL;
 
 	size_t interfacesize = interface_ops->interface_size;
 
@@ -436,9 +411,6 @@ int _starpu_data_handle_init(starpu_data_handle_t handle, struct starpu_data_int
 
 	}
 
-	handle->tag = -1;
-	handle->rank = -1;
-
 	return 0;
 }
 
@@ -492,102 +464,6 @@ void *starpu_data_get_local_ptr(starpu_data_handle_t handle)
 					_starpu_memory_node_get_local_key());
 }
 
-int starpu_data_get_rank(starpu_data_handle_t handle)
-{
-	return handle->rank;
-}
-
-int _starpu_data_set_rank(starpu_data_handle_t handle, int rank)
-{
-	handle->rank = rank;
-	return 0;
-}
-
-int starpu_data_set_rank(starpu_data_handle_t handle, int rank)
-{
-	static int first=1;
-	if (first)
-	{
-		_STARPU_DISP("Warning: You should call starpu_mpi_data_register which will insure MPI cache will be cleared when unregistering the data\n");
-		first=0;
-	}
-	return _starpu_data_set_rank(handle, rank);
-}
-
-int starpu_data_get_tag(starpu_data_handle_t handle)
-{
-	return handle->tag;
-}
-
-starpu_data_handle_t _starpu_data_get_data_handle_from_tag(int tag)
-{
-	struct handle_tag_entry *ret;
-
-	_starpu_spin_lock(&registered_tag_handles_lock);
-	HASH_FIND_INT(registered_tag_handles, &tag, ret);
-	_starpu_spin_unlock(&registered_tag_handles_lock);
-
-	if (ret)
-	{
-		return ret->handle;
-	}
-	else
-	{
-		return NULL;
-	}
-}
-
-int _starpu_data_set_tag(starpu_data_handle_t handle, int tag)
-{
-	struct handle_tag_entry *entry;
-	entry = (struct handle_tag_entry *) malloc(sizeof(*entry));
-	STARPU_ASSERT(entry != NULL);
-
-	STARPU_ASSERT_MSG(!(_starpu_data_get_data_handle_from_tag(tag)),
-			  "There is already a data handle %p registered with the tag %d\n", _starpu_data_get_data_handle_from_tag(tag), tag);
-
-	entry->tag = tag;
-	entry->handle = handle;
-
-	_starpu_spin_lock(&registered_tag_handles_lock);
-	HASH_ADD_INT(registered_tag_handles, tag, entry);
-	_starpu_spin_unlock(&registered_tag_handles_lock);
-
-	handle->tag = tag;
-	return 0;
-}
-
-int starpu_data_set_tag(starpu_data_handle_t handle, int tag)
-{
-	static int first=1;
-	if (first)
-	{
-		_STARPU_DISP("Warning: You should call starpu_mpi_data_register which will insure MPI cache will be cleared when unregistering the data\n");
-		first=0;
-	}
-	return _starpu_data_set_tag(handle, tag);
-}
-
-static
-int _starpu_data_release_tag(starpu_data_handle_t handle)
-{
-	struct handle_tag_entry *tag_entry;
-
-	if (handle->tag != -1)
-	{
-		_starpu_spin_lock(&registered_tag_handles_lock);
-		HASH_FIND_INT(registered_tag_handles, &handle->tag, tag_entry);
-		STARPU_ASSERT_MSG((tag_entry != NULL),"Data handle %p with tag %d isn't in the hashmap !",handle,handle->tag);
-
-		HASH_DEL(registered_tag_handles, tag_entry);
-
-		_starpu_spin_unlock(&registered_tag_handles_lock);
-
-		free(tag_entry);
-	}
-	return 0;
-}
-
 struct starpu_data_interface_ops* starpu_data_get_interface_ops(starpu_data_handle_t handle)
 {
 	return handle->ops;
@@ -925,8 +801,6 @@ static void _starpu_data_unregister(starpu_data_handle_t handle, unsigned cohere
 	STARPU_PTHREAD_COND_DESTROY(&handle->busy_cond);
 	STARPU_PTHREAD_MUTEX_DESTROY(&handle->sequential_consistency_mutex);
 
-	_starpu_data_release_tag(handle);
-
 	free(handle);
 }
 

src/datawizard/interfaces/data_interface.h

@@ -1,7 +1,7 @@
 /* StarPU --- Runtime system for heterogeneous multicore architectures.
  *
  * Copyright (C) 2009-2012, 2014  Université de Bordeaux
- * Copyright (C) 2010, 2012, 2013, 2014  Centre National de la Recherche Scientifique
+ * Copyright (C) 2010, 2012, 2013, 2014, 2015  Centre National de la Recherche Scientifique
  * Copyright (C) 2014  Inria
  *
  * StarPU is free software; you can redistribute it and/or modify
@@ -76,9 +76,5 @@ extern void _starpu_data_unregister_ram_pointer(starpu_data_handle_t handle)
 	STARPU_ATTRIBUTE_INTERNAL;
 
 #define _starpu_data_is_multiformat_handle(handle) handle->ops->is_multiformat
-extern starpu_data_handle_t _starpu_data_get_data_handle_from_tag(int tag);
-
-extern int _starpu_data_set_rank(starpu_data_handle_t handle, int rank);
-extern int _starpu_data_set_tag(starpu_data_handle_t handle, int tag);
 
 #endif // __DATA_INTERFACE_H__