mpi_reduction.c 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2013, 2015, 2017 Université de Bordeaux
  4. * Copyright (C) 2012, 2013, 2014, 2015, 2016, 2017 CNRS
  5. *
  6. * StarPU is free software; you can redistribute it and/or modify
  7. * it under the terms of the GNU Lesser General Public License as published by
  8. * the Free Software Foundation; either version 2.1 of the License, or (at
  9. * your option) any later version.
  10. *
  11. * StarPU is distributed in the hope that it will be useful, but
  12. * WITHOUT ANY WARRANTY; without even the implied warranty of
  13. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  14. *
  15. * See the GNU Lesser General Public License in COPYING.LGPL for more details.
  16. */
  17. #include <starpu_mpi.h>
  18. #include <math.h>
  19. #include "helper.h"
  20. extern void init_cpu_func(void *descr[], void *cl_arg);
  21. extern void redux_cpu_func(void *descr[], void *cl_arg);
  22. extern void dot_cpu_func(void *descr[], void *cl_arg);
  23. extern void display_cpu_func(void *descr[], void *cl_arg);
  24. static struct starpu_codelet init_codelet =
  25. {
  26. .cpu_funcs = {init_cpu_func},
  27. .nbuffers = 1,
  28. .modes = {STARPU_W},
  29. #ifdef STARPU_SIMGRID
  30. .model = &starpu_nop_perf_model,
  31. #endif
  32. .name = "init_codelet"
  33. };
  34. static struct starpu_codelet redux_codelet =
  35. {
  36. .cpu_funcs = {redux_cpu_func},
  37. .modes = {STARPU_RW, STARPU_R},
  38. .nbuffers = 2,
  39. #ifdef STARPU_SIMGRID
  40. .model = &starpu_nop_perf_model,
  41. #endif
  42. .name = "redux_codelet"
  43. };
  44. static struct starpu_codelet dot_codelet =
  45. {
  46. .cpu_funcs = {dot_cpu_func},
  47. .nbuffers = 2,
  48. .modes = {STARPU_R, STARPU_REDUX},
  49. #ifdef STARPU_SIMGRID
  50. .model = &starpu_nop_perf_model,
  51. #endif
  52. .name = "dot_codelet"
  53. };
  54. static struct starpu_codelet display_codelet =
  55. {
  56. .cpu_funcs = {display_cpu_func},
  57. .nbuffers = 1,
  58. .modes = {STARPU_R},
  59. #ifdef STARPU_SIMGRID
  60. .model = &starpu_nop_perf_model,
  61. #endif
  62. .name = "display_codelet"
  63. };
  64. /* Returns the MPI node number where data indexes index is */
  65. int my_distrib(int x, int nb_nodes)
  66. {
  67. return x % nb_nodes;
  68. }
  69. int main(int argc, char **argv)
  70. {
  71. int my_rank, size, x, y, i;
  72. long int *vector;
  73. long int dot, sum=0;
  74. starpu_data_handle_t *handles;
  75. starpu_data_handle_t dot_handle;
  76. int nb_elements, step, loops;
  77. STARPU_SKIP_IF_VALGRIND_RETURN_SKIP;
  78. /* Not supported yet */
  79. if (starpu_get_env_number_default("STARPU_GLOBAL_ARBITER", 0) > 0)
  80. return STARPU_TEST_SKIPPED;
  81. int ret = starpu_init(NULL);
  82. STARPU_CHECK_RETURN_VALUE(ret, "starpu_init");
  83. ret = starpu_mpi_init(&argc, &argv, 1);
  84. STARPU_CHECK_RETURN_VALUE(ret, "starpu_mpi_init");
  85. starpu_mpi_comm_rank(MPI_COMM_WORLD, &my_rank);
  86. starpu_mpi_comm_size(MPI_COMM_WORLD, &size);
  87. if (starpu_cpu_worker_get_count() == 0)
  88. {
  89. if (my_rank == 0)
  90. FPRINTF(stderr, "We need at least 1 CPU worker.\n");
  91. starpu_mpi_shutdown();
  92. starpu_shutdown();
  93. return STARPU_TEST_SKIPPED;
  94. }
  95. nb_elements = size*8000;
  96. step = 4;
  97. loops = 5;
  98. vector = (long int *) malloc(nb_elements*sizeof(vector[0]));
  99. for(x = 0; x < nb_elements; x+=step)
  100. {
  101. int mpi_rank = my_distrib(x/step, size);
  102. if (mpi_rank == my_rank)
  103. {
  104. for(y=0 ; y<step ; y++)
  105. {
  106. vector[x+y] = x+y+1;
  107. }
  108. }
  109. }
  110. if (my_rank == 0)
  111. {
  112. dot = 14;
  113. sum = (nb_elements * (nb_elements + 1)) / 2;
  114. sum *= loops;
  115. sum += dot;
  116. starpu_variable_data_register(&dot_handle, STARPU_MAIN_RAM, (uintptr_t)&dot, sizeof(dot));
  117. }
  118. else
  119. {
  120. starpu_variable_data_register(&dot_handle, -1, (uintptr_t)NULL, sizeof(dot));
  121. }
  122. handles = (starpu_data_handle_t *) malloc(nb_elements*sizeof(handles[0]));
  123. for(x = 0; x < nb_elements; x+=step)
  124. {
  125. handles[x] = NULL;
  126. int mpi_rank = my_distrib(x/step, size);
  127. if (mpi_rank == my_rank)
  128. {
  129. /* Owning data */
  130. starpu_vector_data_register(&handles[x], STARPU_MAIN_RAM, (uintptr_t)&(vector[x]), step, sizeof(vector[0]));
  131. }
  132. else
  133. {
  134. starpu_vector_data_register(&handles[x], -1, (uintptr_t)NULL, step, sizeof(vector[0]));
  135. }
  136. if (handles[x])
  137. {
  138. starpu_mpi_data_register(handles[x], x, mpi_rank);
  139. }
  140. }
  141. starpu_mpi_data_register(dot_handle, nb_elements+1, 0);
  142. starpu_data_set_reduction_methods(dot_handle, &redux_codelet, &init_codelet);
  143. for (i = 0; i < loops; i++)
  144. {
  145. for (x = 0; x < nb_elements; x+=step)
  146. {
  147. starpu_mpi_task_insert(MPI_COMM_WORLD,
  148. &dot_codelet,
  149. STARPU_R, handles[x],
  150. STARPU_REDUX, dot_handle,
  151. 0);
  152. }
  153. starpu_mpi_redux_data(MPI_COMM_WORLD, dot_handle);
  154. starpu_mpi_task_insert(MPI_COMM_WORLD, &display_codelet, STARPU_R, dot_handle, 0);
  155. }
  156. FPRINTF_MPI(stderr, "Waiting ...\n");
  157. starpu_task_wait_for_all();
  158. for(x = 0; x < nb_elements; x+=step)
  159. {
  160. if (handles[x])
  161. starpu_data_unregister(handles[x]);
  162. }
  163. starpu_data_unregister(dot_handle);
  164. free(vector);
  165. free(handles);
  166. starpu_mpi_shutdown();
  167. starpu_shutdown();
  168. if (my_rank == 0)
  169. {
  170. FPRINTF(stderr, "[%d] sum=%ld\n", my_rank, sum);
  171. }
  172. #ifndef STARPU_SIMGRID
  173. if (my_rank == 0)
  174. {
  175. FPRINTF(stderr, "[%d] dot=%ld\n", my_rank, dot);
  176. FPRINTF(stderr, "%s when computing reduction\n", (sum == dot) ? "Success" : "Error");
  177. if (sum != dot)
  178. return 1;
  179. }
  180. #endif
  181. return 0;
  182. }