mpi: new functions to extract the lower level hashtable for early_data and early_requests

mpi/src/mpi/starpu_mpi_early_data.c

@@ -20,19 +20,11 @@
 #include <mpi/starpu_mpi_early_data.h>
 #include <mpi/starpu_mpi_mpi_backend.h>
 #include <starpu_mpi_private.h>
-#include <common/uthash.h>
 
 #ifdef STARPU_USE_MPI_MPI
 
 /** the hashlist is on 2 levels, the first top level is indexed on (node, rank), the second lower level is indexed on the data tag */
 
-struct _starpu_mpi_early_data_handle_tag_hashlist
-{
-	struct _starpu_mpi_early_data_handle_list list;
-	UT_hash_handle hh;
-	starpu_mpi_tag_t data_tag;
-};
-
 struct _starpu_mpi_early_data_handle_hashlist
 {
 	struct _starpu_mpi_early_data_handle_tag_hashlist *datahash;
@@ -74,9 +66,11 @@ void _starpu_mpi_early_data_shutdown(void)
 	struct _starpu_mpi_early_data_handle_hashlist *current=NULL, *tmp=NULL;
 	HASH_ITER(hh, _starpu_mpi_early_data_handle_hashmap, current, tmp)
 	{
+		_STARPU_MPI_DEBUG(600, "Hash early_data with comm %ld source %d\n", (long int) current->node.comm, current->node.rank);
 		struct _starpu_mpi_early_data_handle_tag_hashlist *tag_entry=NULL, *tag_tmp=NULL;
 		HASH_ITER(hh, current->datahash, tag_entry, tag_tmp)
 		{
+			_STARPU_MPI_DEBUG(600, "Hash 2nd level with tag %ld\n", tag_entry->data_tag);
 			STARPU_ASSERT(_starpu_mpi_early_data_handle_list_empty(&tag_entry->list));
 			HASH_DEL(current->datahash, tag_entry);
 			free(tag_entry);
@@ -106,10 +100,11 @@ struct _starpu_mpi_early_data_handle *_starpu_mpi_early_data_find(struct _starpu
 	struct _starpu_mpi_early_data_handle *early_data_handle;
 
 	STARPU_PTHREAD_MUTEX_LOCK(&_starpu_mpi_early_data_handle_mutex);
-	_STARPU_MPI_DEBUG(60, "Looking for early_data_handle with comm %ld source %d tag %ld\n", (long int)node_tag->comm, node_tag->rank, node_tag->data_tag);
+	_STARPU_MPI_DEBUG(60, "Looking for early_data_handle with comm %ld source %d tag %ld\n", (long int)node_tag->node.comm, node_tag->node.rank, node_tag->data_tag);
 	HASH_FIND(hh, _starpu_mpi_early_data_handle_hashmap, &node_tag->node, sizeof(struct _starpu_mpi_node), hashlist);
 	if (hashlist == NULL)
 	{
+		_STARPU_MPI_DEBUG(600, "No entry for (comm %ld, source %d)\n", (long int)node_tag->node.comm, node_tag->node.rank);
 		early_data_handle = NULL;
 	}
 	else
@@ -118,10 +113,12 @@ struct _starpu_mpi_early_data_handle *_starpu_mpi_early_data_find(struct _starpu
 		HASH_FIND(hh, hashlist->datahash, &node_tag->data_tag, sizeof(starpu_mpi_tag_t), tag_hashlist);
 		if (tag_hashlist == NULL)
 		{
+			_STARPU_MPI_DEBUG(600, "No entry for tag %ld\n", node_tag->data_tag);
 			early_data_handle = NULL;
 		}
 		else if (_starpu_mpi_early_data_handle_list_empty(&tag_hashlist->list))
 		{
+			_STARPU_MPI_DEBUG(600, "List empty for tag %ld\n", node_tag->data_tag);
 			early_data_handle = NULL;
 		}
 		else
@@ -130,16 +127,38 @@ struct _starpu_mpi_early_data_handle *_starpu_mpi_early_data_find(struct _starpu
 			early_data_handle = _starpu_mpi_early_data_handle_list_pop_front(&tag_hashlist->list);
 		}
 	}
-	_STARPU_MPI_DEBUG(60, "Found early_data_handle %p with comm %ld source %d tag %ld\n", early_data_handle, (long int)node_tag->comm, node_tag->rank, node_tag->data_tag);
+	_STARPU_MPI_DEBUG(60, "Found early_data_handle %p with comm %ld source %d tag %ld\n", early_data_handle, (long int)node_tag->node.comm, node_tag->node.rank, node_tag->data_tag);
 	STARPU_PTHREAD_MUTEX_UNLOCK(&_starpu_mpi_early_data_handle_mutex);
 	return early_data_handle;
 }
 
+struct _starpu_mpi_early_data_handle_tag_hashlist *_starpu_mpi_early_data_extract(struct _starpu_mpi_node_tag *node_tag)
+{
+	struct _starpu_mpi_early_data_handle_hashlist *hashlist;
+	struct _starpu_mpi_early_data_handle_tag_hashlist *tag_hashlist = NULL;
+
+	STARPU_PTHREAD_MUTEX_LOCK(&_starpu_mpi_early_data_handle_mutex);
+	_STARPU_MPI_DEBUG(60, "Looking for hashlist for (comm %ld, source %d)\n", (long int)node_tag->node.comm, node_tag->node.rank);
+	HASH_FIND(hh, _starpu_mpi_early_data_handle_hashmap, &node_tag->node, sizeof(struct _starpu_mpi_node), hashlist);
+	if (hashlist)
+	{
+		_STARPU_MPI_DEBUG(60, "Looking for hashlist for (tag %ld)\n", node_tag->data_tag);
+		HASH_FIND(hh, hashlist->datahash, &node_tag->data_tag, sizeof(starpu_mpi_tag_t), tag_hashlist);
+		if (tag_hashlist)
+		{
+			_starpu_mpi_early_data_handle_hashmap_count -= _starpu_mpi_early_data_handle_list_size(&tag_hashlist->list);
+			HASH_DEL(hashlist->datahash, tag_hashlist);
+		}
+	}
+	_STARPU_MPI_DEBUG(60, "Found hashlist %p for (comm %ld, source %d) and (tag %ld)\n", tag_hashlist, (long int)node_tag->node.comm, node_tag->node.rank, node_tag->data_tag);
+	STARPU_PTHREAD_MUTEX_UNLOCK(&_starpu_mpi_early_data_handle_mutex);
+	return tag_hashlist;
+}
+
 void _starpu_mpi_early_data_add(struct _starpu_mpi_early_data_handle *early_data_handle)
 {
 	STARPU_PTHREAD_MUTEX_LOCK(&_starpu_mpi_early_data_handle_mutex);
-	_STARPU_MPI_DEBUG(60, "Trying to add early_data_handle %p with comm %ld source %d tag %ld\n", early_data_handle, (long int)early_data_handle->node_tag.node.comm,
-			  early_data_handle->node_tag.node.rank, early_data_handle->node_tag.data_tag);
+	_STARPU_MPI_DEBUG(60, "Adding early_data_handle %p with comm %ld source %d tag %ld (%p)\n", early_data_handle, (long int)early_data_handle->node_tag.node.comm, early_data_handle->node_tag.node.rank, early_data_handle->node_tag.data_tag, &early_data_handle->node_tag.node);
 
 	struct _starpu_mpi_early_data_handle_hashlist *hashlist;
 	HASH_FIND(hh, _starpu_mpi_early_data_handle_hashmap, &early_data_handle->node_tag.node, sizeof(struct _starpu_mpi_node), hashlist);

mpi/src/mpi/starpu_mpi_early_data.h

@@ -1,6 +1,6 @@
 /* StarPU --- Runtime system for heterogeneous multicore architectures.
  *
- * Copyright (C) 2010-2017                                CNRS
+ * Copyright (C) 2010-2017,2019                           CNRS
  * Copyright (C) 2009-2014,2016                           Université de Bordeaux
  *
  * StarPU is free software; you can redistribute it and/or modify
@@ -23,6 +23,7 @@
 #include <mpi.h>
 #include <common/config.h>
 #include <common/list.h>
+#include <common/uthash.h>
 #include <starpu_mpi_private.h>
 
 #ifdef STARPU_USE_MPI_MPI
@@ -43,6 +44,13 @@ LIST_TYPE(_starpu_mpi_early_data_handle,
 	  starpu_pthread_cond_t req_cond;
 );
 
+struct _starpu_mpi_early_data_handle_tag_hashlist
+{
+	struct _starpu_mpi_early_data_handle_list list;
+	UT_hash_handle hh;
+	starpu_mpi_tag_t data_tag;
+};
+
 void _starpu_mpi_early_data_init(void);
 void _starpu_mpi_early_data_check_termination(void);
 void _starpu_mpi_early_data_shutdown(void);
@@ -51,6 +59,8 @@ struct _starpu_mpi_early_data_handle *_starpu_mpi_early_data_create(struct _star
 struct _starpu_mpi_early_data_handle *_starpu_mpi_early_data_find(struct _starpu_mpi_node_tag *node_tag);
 void _starpu_mpi_early_data_add(struct _starpu_mpi_early_data_handle *early_data_handle);
 
+struct _starpu_mpi_early_data_handle_tag_hashlist *_starpu_mpi_early_data_extract(struct _starpu_mpi_node_tag *node_tag);
+
 #ifdef __cplusplus
 }
 #endif

mpi/src/mpi/starpu_mpi_early_request.c

@@ -25,13 +25,6 @@
 
 /** stores application requests for which data have not been received yet */
 /** the hashlist is on 2 levels, the first top level is indexed on (node, rank), the second lower level is indexed on the data tag */
-struct _starpu_mpi_early_request_tag_hashlist
-{
-	struct _starpu_mpi_req_list list;
-	UT_hash_handle hh;
-	starpu_mpi_tag_t data_tag;
-};
-
 struct _starpu_mpi_early_request_hashlist
 {
 	struct _starpu_mpi_early_request_tag_hashlist *datahash;
@@ -120,6 +113,34 @@ struct _starpu_mpi_req* _starpu_mpi_early_request_dequeue(starpu_mpi_tag_t data_
 	return found;
 }
 
+struct _starpu_mpi_early_request_tag_hashlist *_starpu_mpi_early_request_extract(starpu_mpi_tag_t data_tag, int source, MPI_Comm comm)
+{
+	struct _starpu_mpi_node_tag node_tag;
+	struct _starpu_mpi_early_request_hashlist *hashlist;
+	struct _starpu_mpi_early_request_tag_hashlist *tag_hashlist = NULL;
+
+	memset(&node_tag, 0, sizeof(struct _starpu_mpi_node_tag));
+	node_tag.node.comm = comm;
+	node_tag.node.rank = source;
+	node_tag.data_tag = data_tag;
+
+	STARPU_PTHREAD_MUTEX_LOCK(&_starpu_mpi_early_request_mutex);
+	_STARPU_MPI_DEBUG(100, "Looking for early_request with comm %ld source %d tag %ld\n", (long int)node_tag.node.comm, node_tag.node.rank, node_tag.data_tag);
+	HASH_FIND(hh, _starpu_mpi_early_request_hash, &node_tag.node, sizeof(struct _starpu_mpi_node), hashlist);
+	if (hashlist)
+	{
+		HASH_FIND(hh, hashlist->datahash, &node_tag.data_tag, sizeof(starpu_mpi_tag_t), tag_hashlist);
+		if (tag_hashlist)
+		{
+			_starpu_mpi_early_request_hash_count -= _starpu_mpi_req_list_size(&tag_hashlist->list);
+			HASH_DEL(hashlist->datahash, tag_hashlist);
+		}
+	}
+	_STARPU_MPI_DEBUG(100, "Found hashlist %p with comm %ld source %d tag %ld\n", hashlist, (long int)node_tag.node.comm, node_tag.node.rank, node_tag.data_tag);
+	STARPU_PTHREAD_MUTEX_UNLOCK(&_starpu_mpi_early_request_mutex);
+	return tag_hashlist;
+}
+
 void _starpu_mpi_early_request_enqueue(struct _starpu_mpi_req *req)
 {
 	STARPU_PTHREAD_MUTEX_LOCK(&_starpu_mpi_early_request_mutex);

mpi/src/mpi/starpu_mpi_early_request.h

@@ -1,6 +1,6 @@
 /* StarPU --- Runtime system for heterogeneous multicore architectures.
  *
- * Copyright (C) 2010-2017                                CNRS
+ * Copyright (C) 2010-2017,2019                           CNRS
  * Copyright (C) 2009-2014                                Université de Bordeaux
  *
  * StarPU is free software; you can redistribute it and/or modify
@@ -31,6 +31,13 @@ extern "C"
 {
 #endif
 
+struct _starpu_mpi_early_request_tag_hashlist
+{
+	struct _starpu_mpi_req_list list;
+	UT_hash_handle hh;
+	starpu_mpi_tag_t data_tag;
+};
+
 void _starpu_mpi_early_request_init(void);
 void _starpu_mpi_early_request_shutdown(void);
 int _starpu_mpi_early_request_count(void);
@@ -39,6 +46,8 @@ void _starpu_mpi_early_request_check_termination(void);
 void _starpu_mpi_early_request_enqueue(struct _starpu_mpi_req *req);
 struct _starpu_mpi_req* _starpu_mpi_early_request_dequeue(starpu_mpi_tag_t data_tag, int source, MPI_Comm comm);
 
+struct _starpu_mpi_early_request_tag_hashlist *_starpu_mpi_early_request_extract(starpu_mpi_tag_t data_tag, int source, MPI_Comm comm);
+
 #ifdef __cplusplus
 }
 #endif

mpi/src/mpi/starpu_mpi_mpi.c

@@ -1327,7 +1327,7 @@ static void *_starpu_mpi_progress_thread_func(void *arg)
 					{
 						if (envelope->sync)
 						{
-							_STARPU_MPI_DEBUG(2000, "-------------------------> adding request for tag %l"PRIi64"\n", envelope->data_tag);
+							_STARPU_MPI_DEBUG(2000, "-------------------------> adding request for tag %"PRIi64"\n", envelope->data_tag);
 							struct _starpu_mpi_req *new_req;
 #ifdef STARPU_DEVEL
 #warning creating a request is not really useful.

mpi/tests/Makefile.am

@@ -118,7 +118,8 @@ starpu_mpi_TESTS +=				\
 	policy_selection			\
 	policy_selection2			\
 	ring_async_implicit			\
-	temporary
+	temporary				\
+	early_stuff
 
 if !STARPU_SIMGRID
 starpu_mpi_TESTS +=				\
@@ -183,6 +184,7 @@ noinst_PROGRAMS =				\
 	ring_async				\
 	ring_async_implicit			\
 	temporary				\
+	early_stuff				\
 	block_interface				\
 	block_interface_pinned			\
 	attr					\

mpi/tests/early_stuff.c

@@ -0,0 +1,119 @@
+/* StarPU --- Runtime system for heterogeneous multicore architectures.
+ *
+ * Copyright (C) 2019                                     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_mpi.h>
+#include <math.h>
+#include "helper.h"
+
+#ifndef STARPU_USE_MPI_MPI
+int main(int argc, char **argv)
+{
+	return STARPU_TEST_SKIPPED;
+}
+
+#else
+
+#include <mpi/starpu_mpi_early_data.h>
+#include <mpi/starpu_mpi_early_request.h>
+#include <mpi/starpu_mpi_mpi_backend.h>
+
+void early_data()
+{
+	struct _starpu_mpi_early_data_handle *edh[2];
+	struct _starpu_mpi_envelope envelope[2];
+	struct _starpu_mpi_node_tag node_tag[2];
+	struct _starpu_mpi_early_data_handle *early;
+	struct _starpu_mpi_early_data_handle_tag_hashlist *hash;
+
+	memset(&node_tag[0], 0, sizeof(struct _starpu_mpi_node_tag));
+	node_tag[0].node.rank = 1;
+	node_tag[0].node.comm = MPI_COMM_WORLD;
+	node_tag[0].data_tag = 42;
+
+	memset(&node_tag[1], 0, sizeof(struct _starpu_mpi_node_tag));
+	node_tag[1].node.rank = 2;
+	node_tag[1].node.comm = MPI_COMM_WORLD;
+	node_tag[1].data_tag = 84;
+
+	envelope[0].data_tag = node_tag[0].data_tag;
+	edh[0] = _starpu_mpi_early_data_create(&envelope[0], node_tag[0].node.rank, node_tag[0].node.comm);
+
+	envelope[1].data_tag = node_tag[1].data_tag;
+	edh[1] = _starpu_mpi_early_data_create(&envelope[1], node_tag[1].node.rank, node_tag[1].node.comm);
+
+	_starpu_mpi_early_data_add(edh[0]);
+	_starpu_mpi_early_data_add(edh[1]);
+
+	hash = _starpu_mpi_early_data_extract(&node_tag[1]);
+	STARPU_ASSERT(_starpu_mpi_early_data_handle_list_size(&hash->list) == 1);
+	early = _starpu_mpi_early_data_handle_list_pop_front(&hash->list);
+	STARPU_ASSERT(early->node_tag.node.comm == node_tag[1].node.comm && early->node_tag.node.rank == node_tag[1].node.rank && early->node_tag.data_tag == node_tag[1].data_tag);
+	STARPU_ASSERT(_starpu_mpi_early_data_handle_list_size(&hash->list) == 0);
+
+	early = _starpu_mpi_early_data_find(&node_tag[0]);
+	STARPU_ASSERT(early->node_tag.node.comm == node_tag[0].node.comm && early->node_tag.node.rank == node_tag[0].node.rank && early->node_tag.data_tag == node_tag[0].data_tag);
+}
+
+void early_request()
+{
+	struct _starpu_mpi_req req[2];
+	struct _starpu_mpi_req *early;
+	struct _starpu_mpi_early_request_tag_hashlist *hash;
+
+	memset(&req[0].node_tag, 0, sizeof(struct _starpu_mpi_node_tag));
+	req[0].node_tag.node.rank = 1;
+	req[0].node_tag.node.comm = MPI_COMM_WORLD;
+	req[0].node_tag.data_tag = 42;
+
+	memset(&req[1].node_tag, 0, sizeof(struct _starpu_mpi_node_tag));
+	req[1].node_tag.node.rank = 2;
+	req[1].node_tag.node.comm = MPI_COMM_WORLD;
+	req[1].node_tag.data_tag = 84;
+
+	_starpu_mpi_early_request_enqueue(&req[1]);
+	_starpu_mpi_early_request_enqueue(&req[0]);
+
+	early = _starpu_mpi_early_request_dequeue(req[0].node_tag.data_tag, req[0].node_tag.node.rank, req[0].node_tag.node.comm);
+	STARPU_ASSERT(early->node_tag.data_tag == req[0].node_tag.data_tag && early->node_tag.node.rank == req[0].node_tag.node.rank && early->node_tag.node.comm == req[0].node_tag.node.comm);
+
+	hash = _starpu_mpi_early_request_extract(req[1].node_tag.data_tag, req[1].node_tag.node.rank, req[1].node_tag.node.comm);
+	STARPU_ASSERT(_starpu_mpi_req_list_size(&hash->list) == 1);
+	early = _starpu_mpi_req_list_pop_front(&hash->list);
+	STARPU_ASSERT(_starpu_mpi_req_list_size(&hash->list) == 0);
+	STARPU_ASSERT(early->node_tag.data_tag == req[1].node_tag.data_tag && early->node_tag.node.rank == req[1].node_tag.node.rank && early->node_tag.node.comm == req[1].node_tag.node.comm);
+}
+
+int main(int argc, char **argv)
+{
+	int ret, rank, size, i;
+	starpu_data_handle_t tab_handle[4];
+	int mpi_init;
+
+	MPI_INIT_THREAD(&argc, &argv, MPI_THREAD_SERIALIZED, &mpi_init);
+	ret = starpu_mpi_init_conf(&argc, &argv, mpi_init, MPI_COMM_WORLD, NULL);
+	STARPU_CHECK_RETURN_VALUE(ret, "starpu_mpi_init_conf");
+
+	early_data();
+	early_request();
+
+	starpu_mpi_shutdown();
+
+	if (!mpi_init)
+		MPI_Finalize();
+	return 0;
+}
+
+#endif