mpi: New variable STARPU_MPI_CACHE_STATS to print statistics on cache holding received data.

ChangeLog

@@ -121,6 +121,8 @@ New features:
         - Communication cache mechanism is enabled by default, and can
 	  only be disabled at execution time by setting the
 	  environment variable STARPU_MPI_CACHE to 0.
+        - New variable STARPU_MPI_CACHE_STATS to print statistics on
+   	  cache holding received data.
         - Initialisation functions starpu_mpi_initialize_extended()
   	  and starpu_mpi_initialize() have been made deprecated. One
 	  should now use starpu_mpi_init(int *, char ***, int). The

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 1
- * Copyright (C) 2010, 2011, 2012, 2013  Centre National de la Recherche Scientifique
+ * Copyright (C) 2010, 2011, 2012, 2013, 2014  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.
  */
@@ -363,7 +363,9 @@ the close future. If a newly-submitted task actually needs the value again,
 another transmission of D will be initiated from A to B.
 
 The whole caching behavior can be disabled thanks to the ::STARPU_MPI_CACHE
-environment variable.
+environment variable. The variable ::STARPU_MPI_CACHE_STATS can be set to 1
+to enable the runtime to display messages when data are added or removed
+from the cache holding the received data.
 
 \section MPIMigration MPI Data migration
 

doc/doxygen/chapters/40environment_variables.doxy

@@ -1,7 +1,7 @@
 /*
  * This file is part of the StarPU Handbook.
  * Copyright (C) 2009--2011  Universit@'e de Bordeaux 1
- * Copyright (C) 2010, 2011, 2012, 2013  Centre National de la Recherche Scientifique
+ * Copyright (C) 2010, 2011, 2012, 2013, 2014  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.
  */
@@ -402,6 +402,15 @@ to 0. It is enabled by default or for any other values of the variable
 \ref STARPU_MPI_CACHE.
 </dd>
 
+<dt>STARPU_MPI_CACHE_STATS</dt>
+<dd>
+\anchor STARPU_MPI_CACHE_STATS
+\addindex __env__STARPU_MPI_CACHE_STATS
+When set to 1, statistics are enabled for the communication cache (\ref MPISupport). For now,
+it prints messages on the standard output when data are added or removed from the received
+communication cache.
+</dd>
+
 </dl>
 
 \section MiscellaneousAndDebug Miscellaneous And Debug

mpi/src/Makefile.am

@@ -1,7 +1,7 @@
 # StarPU --- Runtime system for heterogeneous multicore architectures.
 #
 # Copyright (C) 2009-2012  Université de Bordeaux 1
-# Copyright (C) 2010, 2011, 2012, 2013  Centre National de la Recherche Scientifique
+# Copyright (C) 2010, 2011, 2012, 2013, 2014  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
@@ -37,7 +37,8 @@ noinst_HEADERS =					\
 	starpu_mpi_fxt.h				\
 	starpu_mpi_stats.h				\
 	starpu_mpi_task_insert.h			\
-	starpu_mpi_datatype.h
+	starpu_mpi_datatype.h				\
+	starpu_mpi_cache_stats.h
 
 libstarpumpi_@STARPU_EFFECTIVE_VERSION@_la_SOURCES =	\
 	starpu_mpi.c					\
@@ -46,8 +47,8 @@ libstarpumpi_@STARPU_EFFECTIVE_VERSION@_la_SOURCES =	\
 	starpu_mpi_task_insert.c			\
 	starpu_mpi_collective.c				\
 	starpu_mpi_stats.c				\
-	starpu_mpi_private.c
-
+	starpu_mpi_private.c				\
+	starpu_mpi_cache_stats.c
 
 showcheck:
 	-cat /dev/null

mpi/src/starpu_mpi_cache_stats.c

@@ -0,0 +1,70 @@
+/* StarPU --- Runtime system for heterogeneous multicore architectures.
+ *
+ * Copyright (C) 2014  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.
+ */
+
+#include <starpu_mpi_cache_stats.h>
+#include <common/config.h>
+#include <stdio.h>
+#include <starpu_mpi_private.h>
+
+/* measure the amount of data transfers between each pair of MPI nodes */
+static size_t *comm_cache_amount;
+static int world_size;
+static int stats_enabled=0;
+
+void _starpu_mpi_cache_stats_init(MPI_Comm comm)
+{
+	stats_enabled = starpu_get_env_number("STARPU_MPI_CACHE_STATS");
+	if (stats_enabled == -1)
+	{
+		stats_enabled = 0;
+	}
+
+	if (stats_enabled == 0) return;
+
+	if (!getenv("STARPU_SILENT")) fprintf(stderr,"Warning: StarPU is executed with STARPU_MPI_CACHE_STATS=1, which slows down a bit\n");
+
+	MPI_Comm_size(comm, &world_size);
+	_STARPU_MPI_DEBUG(1, "allocating for %d nodes\n", world_size);
+
+	comm_cache_amount = (size_t *) calloc(world_size, sizeof(size_t));
+}
+
+void _starpu_mpi_cache_stats_free()
+{
+	if (stats_enabled == 0) return;
+	free(comm_cache_amount);
+}
+
+void _starpu_mpi_cache_stats_update(int src, unsigned dst, starpu_data_handle_t data_handle, int count)
+{
+	size_t size;
+
+	if (stats_enabled == 0) return;
+
+	size = starpu_data_get_size(data_handle);
+
+	if (count == 1)
+	{
+		fprintf(stderr, "[starpu_mpi_cache_stats][%d] adding %ld to %d\n", src, (long)size, dst);
+	}
+	else // count == -1
+	{
+		fprintf(stderr, "[starpu_mpi_cache_stats][%d] removing %ld from %d\n", src, (long)size, dst);
+	}
+
+	comm_cache_amount[dst] += count * size;
+}
+

mpi/src/starpu_mpi_cache_stats.h

@@ -0,0 +1,40 @@
+/* StarPU --- Runtime system for heterogeneous multicore architectures.
+ *
+ * Copyright (C) 2014  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_CACHE_STATS_H__
+#define __STARPU_MPI_CACHE_STATS_H__
+
+#include <starpu.h>
+#include <stdlib.h>
+#include <mpi.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+void _starpu_mpi_cache_stats_init(MPI_Comm comm);
+void _starpu_mpi_cache_stats_free();
+
+void _starpu_mpi_cache_stats_update(int src, unsigned dst, starpu_data_handle_t data_handle, int count);
+
+#define _starpu_mpi_cache_stats_inc(src, dst, data_handle) _starpu_mpi_cache_stats_update(src, dst, data_handle, +1)
+#define _starpu_mpi_cache_stats_dec(src, dst, data_handle) _starpu_mpi_cache_stats_update(src, dst, data_handle, -1)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // __STARPU_MPI_CACHE_STATS_H__

mpi/src/starpu_mpi_task_insert.c

@@ -27,6 +27,7 @@
 #include <datawizard/coherency.h>
 #include <core/task.h>
 
+#include <starpu_mpi_cache_stats.h>
 #include <starpu_mpi_private.h>
 #include <starpu_mpi_task_insert.h>
 
@@ -64,6 +65,7 @@ void _starpu_mpi_cache_init(MPI_Comm comm)
 	for(i=0 ; i<nb_nodes ; i++) _cache_sent_data[i] = NULL;
 	_cache_received_data = malloc(nb_nodes * sizeof(struct _starpu_data_entry *));
 	for(i=0 ; i<nb_nodes ; i++) _cache_received_data[i] = NULL;
+	_starpu_mpi_cache_stats_init(comm);
 }
 
 static
@@ -86,6 +88,7 @@ void _starpu_mpi_cache_empty_tables(int world_size)
 		HASH_ITER(hh, _cache_received_data[i], entry, tmp)
 		{
 			HASH_DEL(_cache_received_data[i], entry);
+			_starpu_mpi_cache_stats_dec(-1, i, (starpu_data_handle_t) entry->data);
 			free(entry);
 		}
 	}
@@ -127,6 +130,7 @@ void starpu_mpi_cache_flush_all_data(MPI_Comm comm)
 			if (mpi_rank != my_rank && mpi_rank != -1)
 				starpu_data_invalidate_submit((starpu_data_handle_t) entry->data);
 			HASH_DEL(_cache_received_data[i], entry);
+			_starpu_mpi_cache_stats_dec(my_rank, i, (starpu_data_handle_t) entry->data);
 			free(entry);
 		}
 	}
@@ -158,6 +162,7 @@ void starpu_mpi_cache_flush(MPI_Comm comm, starpu_data_handle_t data_handle)
 		{
 			_STARPU_MPI_DEBUG(2, "Clearing send cache for data %p\n", data_handle);
 			HASH_DEL(_cache_received_data[i], avail);
+			_starpu_mpi_cache_stats_dec(my_rank, i, data_handle);
 			free(avail);
 		}
 	}
@@ -167,7 +172,7 @@ void starpu_mpi_cache_flush(MPI_Comm comm, starpu_data_handle_t data_handle)
 }
 
 static
-void *_starpu_mpi_already_received(starpu_data_handle_t data, int mpi_rank)
+void *_starpu_mpi_already_received(int src, starpu_data_handle_t data, int mpi_rank)
 {
 	if (_cache_enabled == 0) return NULL;
 
@@ -178,6 +183,7 @@ void *_starpu_mpi_already_received(starpu_data_handle_t data, int mpi_rank)
 		struct _starpu_data_entry *entry = (struct _starpu_data_entry *)malloc(sizeof(*entry));
 		entry->data = data;
 		HASH_ADD_PTR(_cache_received_data[mpi_rank], data, entry);
+		_starpu_mpi_cache_stats_inc(src, mpi_rank, data);
 	}
 	else
 	{
@@ -287,7 +293,7 @@ void _starpu_mpi_exchange_data_before_execution(starpu_data_handle_t data, enum
 		if (do_execute && mpi_rank != me && mpi_rank != -1)
 		{
 			/* I will have to execute but I don't have the data, receive */
-			void *already_received = _starpu_mpi_already_received(data, mpi_rank);
+			void *already_received = _starpu_mpi_already_received(me, data, mpi_rank);
 			if (already_received == NULL)
 			{
 				_STARPU_MPI_DEBUG(1, "Receive data %p from %d\n", data, mpi_rank);
@@ -376,6 +382,7 @@ void _starpu_mpi_clear_data_after_execution(starpu_data_handle_t data, enum star
 #endif
 					_STARPU_MPI_DEBUG(2, "Clearing receive cache for data %p\n", data);
 					HASH_DEL(_cache_received_data[mpi_rank], already_received);
+					_starpu_mpi_cache_stats_dec(me, mpi_rank, data);
 					free(already_received);
 					starpu_data_invalidate_submit(data);
 				}