mpi/tests: new testcase for cache mechanism

mpi/tests/Makefile.am

@@ -69,7 +69,7 @@ endif
 
 AM_CFLAGS = -Wall $(STARPU_CUDA_CPPFLAGS) $(STARPU_OPENCL_CPPFLAGS) $(FXT_CFLAGS) $(MAGMA_CFLAGS) $(HWLOC_CFLAGS) $(GLOBAL_AM_CFLAGS) -Wno-unused
 LIBS = $(top_builddir)/src/@LIBSTARPU_LINK@ @LIBS@ $(FXT_LIBS) $(MAGMA_LIBS)
-AM_CPPFLAGS = -I$(top_srcdir)/include/ -I$(top_builddir)/include -I$(top_srcdir)/mpi/include -I$(top_srcdir)/src -I$(top_builddir)/src -I$(top_srcdir)/examples/
+AM_CPPFLAGS = -I$(top_srcdir)/include/ -I$(top_builddir)/include -I$(top_srcdir)/mpi/include -I$(top_srcdir)/mpi/src -I$(top_srcdir)/src -I$(top_builddir)/src -I$(top_srcdir)/examples/
 AM_LDFLAGS = $(STARPU_OPENCL_LDFLAGS) $(STARPU_CUDA_LDFLAGS) $(FXT_LDFLAGS) $(STARPU_COI_LDFLAGS) $(STARPU_SCIF_LDFLAGS)
 
 ########################
@@ -95,8 +95,10 @@ starpu_mpi_TESTS =				\
 	ring_async_implicit			\
 	block_interface				\
 	block_interface_pinned			\
+	cache					\
 	callback				\
 	matrix					\
+	matrix2					\
 	insert_task				\
 	insert_task_compute			\
 	insert_task_sent_cache			\
@@ -134,8 +136,10 @@ noinst_PROGRAMS =				\
 	ring_async_implicit			\
 	block_interface				\
 	block_interface_pinned			\
+	cache					\
 	callback				\
 	matrix					\
+	matrix2					\
 	insert_task				\
 	insert_task_compute			\
 	insert_task_sent_cache			\
@@ -190,10 +194,14 @@ block_interface_LDADD =				\
 	../src/libstarpumpi-@STARPU_EFFECTIVE_VERSION@.la
 block_interface_pinned_LDADD =			\
 	../src/libstarpumpi-@STARPU_EFFECTIVE_VERSION@.la
+cache_LDADD =					\
+	../src/libstarpumpi-@STARPU_EFFECTIVE_VERSION@.la
 callback_LDADD =				\
 	../src/libstarpumpi-@STARPU_EFFECTIVE_VERSION@.la
 matrix_LDADD =					\
 	../src/libstarpumpi-@STARPU_EFFECTIVE_VERSION@.la
+matrix2_LDADD =					\
+	../src/libstarpumpi-@STARPU_EFFECTIVE_VERSION@.la
 insert_task_LDADD =				\
 	../src/libstarpumpi-@STARPU_EFFECTIVE_VERSION@.la
 insert_task_compute_LDADD =				\

mpi/tests/cache.c

@@ -0,0 +1,113 @@
+/* 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.
+ */
+
+#include <starpu_mpi.h>
+#include <math.h>
+#include "helper.h"
+#include <starpu_mpi_cache.h>
+
+void func_cpu(STARPU_ATTRIBUTE_UNUSED void *descr[], STARPU_ATTRIBUTE_UNUSED void *_args)
+{
+}
+
+struct starpu_codelet mycodelet_r =
+{
+	.cpu_funcs = {func_cpu},
+	.nbuffers = 1,
+	.modes = {STARPU_R}
+};
+
+struct starpu_codelet mycodelet_w =
+{
+	.cpu_funcs = {func_cpu},
+	.nbuffers = 1,
+	.modes = {STARPU_W}
+};
+
+struct starpu_codelet mycodelet_rw =
+{
+	.cpu_funcs = {func_cpu},
+	.nbuffers = 1,
+	.modes = {STARPU_RW}
+};
+
+void test(struct starpu_codelet *codelet, enum starpu_data_access_mode mode, starpu_data_handle_t data, int rank, int in_cache)
+{
+	void *ptr;
+	int ret;
+
+	// We call starpu_mpi_task_build() and starpu_mpi_task_post_build() instead of
+	// starpu_mpi_task_insert() to avoid executing the codelet as we just want to test the cache mechanism
+	starpu_mpi_task_build(MPI_COMM_WORLD, codelet, mode, data, STARPU_EXECUTE_ON_NODE, 1, 0);
+	ret = starpu_mpi_task_post_build(MPI_COMM_WORLD, codelet, mode, data, STARPU_EXECUTE_ON_NODE, 1, 0);
+	STARPU_CHECK_RETURN_VALUE(ret, "starpu_mpi_task_post_build");
+
+	ptr = _starpu_mpi_cache_received_data_get(data, 0);
+
+	if (rank == 1)
+	{
+	     if (in_cache)
+	     {
+		     STARPU_ASSERT_MSG(ptr != NULL, "Data should be in cache\n");
+	     }
+	     else
+	     {
+		     STARPU_ASSERT_MSG(ptr == NULL, "Data should NOT be in cache\n");
+	     }
+	}
+}
+
+int main(int argc, char **argv)
+{
+	int rank, n;
+	int ret;
+	unsigned val;
+	starpu_data_handle_t data;
+
+	ret = starpu_init(NULL);
+	STARPU_CHECK_RETURN_VALUE(ret, "starpu_init");
+	ret = starpu_mpi_init(&argc, &argv, 1);
+	STARPU_CHECK_RETURN_VALUE(ret, "starpu_mpi_init");
+	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+
+	if (starpu_mpi_cache_is_enabled() == 0) goto skip;
+
+	if (rank == 0)
+		starpu_variable_data_register(&data, STARPU_MAIN_RAM, (uintptr_t)&val, sizeof(unsigned));
+	else
+		starpu_variable_data_register(&data, -1, (uintptr_t)NULL, sizeof(unsigned));
+	starpu_mpi_data_register(data, 42, 0);
+	FPRINTF_MPI(stderr, "Registering data %p with tag %d and node %d\n", data, 42, 0);
+
+	// We use the same data with different access modes and we check if it is
+	// available or not in the cache
+	test(&mycodelet_r, STARPU_R, data, rank, 1);
+	test(&mycodelet_rw, STARPU_RW, data, rank, 0);
+	test(&mycodelet_r, STARPU_R, data, rank, 1);
+	test(&mycodelet_r, STARPU_R, data, rank, 1);
+	test(&mycodelet_w, STARPU_W, data, rank, 0);
+
+	FPRINTF(stderr, "Waiting ...\n");
+	starpu_task_wait_for_all();
+
+	starpu_data_unregister(data);
+
+skip:
+	starpu_mpi_shutdown();
+	starpu_shutdown();
+
+	return starpu_mpi_cache_is_enabled() == 0 ? STARPU_TEST_SKIPPED : 0;
+}

mpi/tests/matrix2.c

@@ -0,0 +1,140 @@
+/* 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.
+ */
+
+#include <starpu_mpi.h>
+#include <math.h>
+#include "helper.h"
+
+void func_cpu(void *descr[], STARPU_ATTRIBUTE_UNUSED void *_args)
+{
+	unsigned *A = (unsigned *)STARPU_VARIABLE_GET_PTR(descr[0]);
+	unsigned *X = (unsigned *)STARPU_VARIABLE_GET_PTR(descr[1]);
+	unsigned *Y = (unsigned *)STARPU_VARIABLE_GET_PTR(descr[2]);
+
+	FPRINTF_MPI(stderr, "VALUES: Y=%3u A=%3u X=%3u\n", *Y, *A, *X);
+	*Y = *Y + *A * *X;
+}
+
+struct starpu_codelet mycodelet =
+{
+	.cpu_funcs = {func_cpu},
+	.nbuffers = 3,
+	.modes = {STARPU_R, STARPU_R, STARPU_RW}
+};
+
+#define N 4
+
+int main(int argc, char **argv)
+{
+	int rank, size;
+	int n;
+	int ret;
+	unsigned A[N];
+	unsigned X[N];
+	starpu_data_handle_t data_A[N];
+	starpu_data_handle_t data_X[N];
+
+	MPI_Init(&argc, &argv);
+	MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+	MPI_Comm_size(MPI_COMM_WORLD, &size);
+
+	if (size < 3)
+	{
+		if (rank == 0)
+			FPRINTF(stderr, "We need at least 3 processes.\n");
+
+		MPI_Finalize();
+		return STARPU_TEST_SKIPPED;
+	}
+
+	ret = starpu_init(NULL);
+	STARPU_CHECK_RETURN_VALUE(ret, "starpu_init");
+	ret = starpu_mpi_init(NULL, NULL, 0);
+	STARPU_CHECK_RETURN_VALUE(ret, "starpu_mpi_init");
+
+	for(n = 0; n < N; n++)
+	{
+		A[n] = (n+1)*10;
+		X[n] = n+1;
+	}
+
+	FPRINTF_MPI(stderr, "A = ");
+	for(n = 0; n < N; n++)
+	{
+		FPRINTF(stderr, "%u ", A[n]);
+	}
+	FPRINTF(stderr, "\n");
+	FPRINTF_MPI(stderr, "X = ");
+	for(n = 0; n < N; n++)
+	{
+		FPRINTF(stderr, "%u ", X[n]);
+	}
+	FPRINTF(stderr, "\n");
+
+	for(n = 0; n < N; n++)
+	{
+		if (rank == n%2)
+			starpu_variable_data_register(&data_A[n], STARPU_MAIN_RAM, (uintptr_t)&A[n], sizeof(unsigned));
+		else
+			starpu_variable_data_register(&data_A[n], -1, (uintptr_t)NULL, sizeof(unsigned));
+		starpu_mpi_data_register(data_A[n], n+100, n%2);
+		FPRINTF_MPI(stderr, "Registering A[%d] to %p with tag %d and node %d\n", n,data_A[n], n+100, n%2);
+	}
+
+	for(n = 0; n < N; n++)
+	{
+		if (rank == 2)
+			starpu_variable_data_register(&data_X[n], STARPU_MAIN_RAM, (uintptr_t)&X[n], sizeof(unsigned));
+		else
+			starpu_variable_data_register(&data_X[n], -1, (uintptr_t)NULL, sizeof(unsigned));
+		starpu_mpi_data_register(data_X[n], n+200, 2);
+		FPRINTF_MPI(stderr, "Registering X[%d] to %p with tag %d and node %d\n", n, data_X[n], n+200, 2);
+	}
+
+	for(n = 0; n < N-1; n++)
+	{
+	     fprintf(stderr, "loop %d\n", n);
+		ret = starpu_mpi_task_insert(MPI_COMM_WORLD, &mycodelet,
+					     STARPU_R, data_A[n],
+					     STARPU_R, data_X[n],
+					     STARPU_RW, data_X[N-1],
+					     STARPU_EXECUTE_ON_DATA, data_A[n],
+					     0);
+		STARPU_CHECK_RETURN_VALUE(ret, "starpu_mpi_task_insert");
+	}
+
+	FPRINTF(stderr, "Waiting ...\n");
+	starpu_task_wait_for_all();
+
+	for(n = 0; n < N; n++)
+	{
+		starpu_data_unregister(data_A[n]);
+		starpu_data_unregister(data_X[n]);
+	}
+
+	starpu_mpi_shutdown();
+	starpu_shutdown();
+
+	FPRINTF(stdout, "[%d] X[%d]=%u\n", rank, N-1, X[N-1]);
+
+	if (rank == 2)
+	{
+		STARPU_ASSERT_MSG(X[N-1]==144, "Error when calculating X[N-1]=%u\n", X[N-1]);
+	}
+
+	MPI_Finalize();
+	return 0;
+}