starpu_mpi_init.c 7.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2016,2017 Inria
  4. * Copyright (C) 2010-2018 CNRS
  5. * Copyright (C) 2009-2018 Université de Bordeaux
  6. *
  7. * StarPU is free software; you can redistribute it and/or modify
  8. * it under the terms of the GNU Lesser General Public License as published by
  9. * the Free Software Foundation; either version 2.1 of the License, or (at
  10. * your option) any later version.
  11. *
  12. * StarPU is distributed in the hope that it will be useful, but
  13. * WITHOUT ANY WARRANTY; without even the implied warranty of
  14. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  15. *
  16. * See the GNU Lesser General Public License in COPYING.LGPL for more details.
  17. */
  18. #include <stdlib.h>
  19. #include <starpu_mpi.h>
  20. #include <starpu_mpi_datatype.h>
  21. #include <starpu_mpi_private.h>
  22. #include <starpu_mpi_cache.h>
  23. #include <starpu_profiling.h>
  24. #include <starpu_mpi_stats.h>
  25. #include <starpu_mpi_cache.h>
  26. #include <starpu_mpi_select_node.h>
  27. #include <common/config.h>
  28. #include <common/thread.h>
  29. #include <datawizard/interfaces/data_interface.h>
  30. #include <datawizard/coherency.h>
  31. #include <core/simgrid.h>
  32. #include <core/task.h>
  33. #if defined(STARPU_USE_MPI_MPI)
  34. #include <mpi/starpu_mpi_comm.h>
  35. #include <mpi/starpu_mpi_tag.h>
  36. #include <mpi/starpu_mpi_driver.h>
  37. #endif
  38. #ifdef STARPU_SIMGRID
  39. static int _mpi_world_size;
  40. static int _mpi_world_rank;
  41. #endif
  42. static int _mpi_initialized_starpu;
  43. static void _starpu_mpi_print_thread_level_support(int thread_level, char *msg)
  44. {
  45. switch (thread_level)
  46. {
  47. case MPI_THREAD_SERIALIZED:
  48. {
  49. _STARPU_DISP("MPI%s MPI_THREAD_SERIALIZED; Multiple threads may make MPI calls, but only one at a time.\n", msg);
  50. break;
  51. }
  52. case MPI_THREAD_FUNNELED:
  53. {
  54. _STARPU_DISP("MPI%s MPI_THREAD_FUNNELED; The application can safely make calls to StarPU-MPI functions, but should not call directly MPI communication functions.\n", msg);
  55. break;
  56. }
  57. case MPI_THREAD_SINGLE:
  58. {
  59. _STARPU_DISP("MPI%s MPI_THREAD_SINGLE; MPI does not have multi-thread support, this might cause problems. The application can make calls to StarPU-MPI functions, but not call directly MPI Communication functions.\n", msg);
  60. break;
  61. }
  62. case MPI_THREAD_MULTIPLE:
  63. /* no problem */
  64. break;
  65. }
  66. }
  67. void _starpu_mpi_do_initialize(struct _starpu_mpi_argc_argv *argc_argv)
  68. {
  69. if (argc_argv->initialize_mpi)
  70. {
  71. STARPU_ASSERT_MSG(argc_argv->comm == MPI_COMM_WORLD, "It does not make sense to ask StarPU-MPI to initialize MPI while a non-world communicator was given");
  72. int thread_support;
  73. #ifdef STARPU_USE_MPI_NMAD
  74. /* strat_prio is preferred for StarPU instead of default strat_aggreg */
  75. setenv("NMAD_STRATEGY", "prio", 0 /* do not overwrite user-supplied value, if set */);
  76. #endif /* STARPU_USE_MPI_NMAD */
  77. _STARPU_DEBUG("Calling MPI_Init_thread\n");
  78. if (MPI_Init_thread(argc_argv->argc, argc_argv->argv, MPI_THREAD_SERIALIZED, &thread_support) != MPI_SUCCESS)
  79. {
  80. _STARPU_ERROR("MPI_Init_thread failed\n");
  81. }
  82. _starpu_mpi_print_thread_level_support(thread_support, "_Init_thread level =");
  83. }
  84. else
  85. {
  86. int provided;
  87. MPI_Query_thread(&provided);
  88. _starpu_mpi_print_thread_level_support(provided, " has been initialized with");
  89. }
  90. MPI_Comm_rank(argc_argv->comm, &argc_argv->rank);
  91. MPI_Comm_size(argc_argv->comm, &argc_argv->world_size);
  92. MPI_Comm_set_errhandler(argc_argv->comm, MPI_ERRORS_RETURN);
  93. #ifdef STARPU_SIMGRID
  94. _mpi_world_size = argc_argv->world_size;
  95. _mpi_world_rank = argc_argv->rank;
  96. #endif
  97. }
  98. static
  99. int _starpu_mpi_initialize(int *argc, char ***argv, int initialize_mpi, MPI_Comm comm)
  100. {
  101. struct _starpu_mpi_argc_argv *argc_argv;
  102. _STARPU_MALLOC(argc_argv, sizeof(struct _starpu_mpi_argc_argv));
  103. argc_argv->initialize_mpi = initialize_mpi;
  104. argc_argv->argc = argc;
  105. argc_argv->argv = argv;
  106. argc_argv->comm = comm;
  107. _starpu_implicit_data_deps_write_hook(_starpu_mpi_data_flush);
  108. #ifdef STARPU_SIMGRID
  109. /* Call MPI_Init_thread as early as possible, to initialize simgrid
  110. * before working with mutexes etc. */
  111. _starpu_mpi_do_initialize(argc_argv);
  112. #endif
  113. return _starpu_mpi_progress_init(argc_argv);
  114. }
  115. #ifdef STARPU_SIMGRID
  116. /* This is called before application's main, to initialize SMPI before we can
  117. * create MSG processes to run application's main */
  118. int _starpu_mpi_simgrid_init(int argc, char *argv[])
  119. {
  120. return _starpu_mpi_initialize(&argc, &argv, 1, MPI_COMM_WORLD);
  121. }
  122. #endif
  123. int starpu_mpi_init_comm(int *argc, char ***argv, int initialize_mpi, MPI_Comm comm)
  124. {
  125. #ifdef STARPU_SIMGRID
  126. (void)argc;
  127. (void)argv;
  128. (void)initialize_mpi;
  129. (void)comm;
  130. _starpu_mpi_wait_for_initialization();
  131. return 0;
  132. #else
  133. return _starpu_mpi_initialize(argc, argv, initialize_mpi, comm);
  134. #endif
  135. }
  136. int starpu_mpi_init(int *argc, char ***argv, int initialize_mpi)
  137. {
  138. return starpu_mpi_init_comm(argc, argv, initialize_mpi, MPI_COMM_WORLD);
  139. }
  140. int starpu_mpi_initialize(void)
  141. {
  142. #ifdef STARPU_SIMGRID
  143. return 0;
  144. #else
  145. return _starpu_mpi_initialize(NULL, NULL, 0, MPI_COMM_WORLD);
  146. #endif
  147. }
  148. int starpu_mpi_initialize_extended(int *rank, int *world_size)
  149. {
  150. #ifdef STARPU_SIMGRID
  151. *world_size = _mpi_world_size;
  152. *rank = _mpi_world_rank;
  153. return 0;
  154. #else
  155. int ret;
  156. ret = _starpu_mpi_initialize(NULL, NULL, 1, MPI_COMM_WORLD);
  157. if (ret == 0)
  158. {
  159. _STARPU_DEBUG("Calling MPI_Comm_rank\n");
  160. MPI_Comm_rank(MPI_COMM_WORLD, rank);
  161. MPI_Comm_size(MPI_COMM_WORLD, world_size);
  162. }
  163. return ret;
  164. #endif
  165. }
  166. int starpu_mpi_init_conf(int *argc, char ***argv, int initialize_mpi, MPI_Comm comm, struct starpu_conf *conf)
  167. {
  168. struct starpu_conf localconf;
  169. if (!conf)
  170. {
  171. starpu_conf_init(&localconf);
  172. conf = &localconf;
  173. }
  174. #if defined(STARPU_USE_MPI_MPI)
  175. _starpu_mpi_driver_init(conf);
  176. if (starpu_get_env_number_default("STARPU_MPI_DRIVER_CALL_FREQUENCY", 0) <= 0)
  177. #endif
  178. {
  179. /* Reserve a core for our progression thread */
  180. if (conf->reserve_ncpus == -1)
  181. conf->reserve_ncpus = 1;
  182. else
  183. conf->reserve_ncpus++;
  184. }
  185. int ret = starpu_init(conf);
  186. if (ret < 0)
  187. return ret;
  188. _mpi_initialized_starpu = 1;
  189. return starpu_mpi_init_comm(argc, argv, initialize_mpi, comm);
  190. }
  191. int starpu_mpi_shutdown(void)
  192. {
  193. void *value;
  194. int rank, world_size;
  195. /* We need to get the rank before calling MPI_Finalize to pass to _starpu_mpi_comm_amounts_display() */
  196. starpu_mpi_comm_rank(MPI_COMM_WORLD, &rank);
  197. starpu_mpi_comm_size(MPI_COMM_WORLD, &world_size);
  198. /* kill the progression thread */
  199. _starpu_mpi_progress_shutdown(&value);
  200. _STARPU_MPI_TRACE_STOP(rank, world_size);
  201. _starpu_mpi_comm_amounts_display(stderr, rank);
  202. _starpu_mpi_comm_amounts_shutdown();
  203. _starpu_mpi_cache_shutdown(world_size);
  204. #if defined(STARPU_USE_MPI_MPI)
  205. _starpu_mpi_tag_shutdown();
  206. _starpu_mpi_comm_shutdown();
  207. _starpu_mpi_driver_shutdown();
  208. #endif
  209. if (_mpi_initialized_starpu)
  210. starpu_shutdown();
  211. return 0;
  212. }
  213. int starpu_mpi_comm_size(MPI_Comm comm, int *size)
  214. {
  215. if (_starpu_mpi_fake_world_size != -1)
  216. {
  217. *size = _starpu_mpi_fake_world_size;
  218. return 0;
  219. }
  220. #ifdef STARPU_SIMGRID
  221. STARPU_MPI_ASSERT_MSG(comm == MPI_COMM_WORLD, "StarPU-SMPI only works with MPI_COMM_WORLD for now");
  222. *size = _mpi_world_size;
  223. return 0;
  224. #else
  225. return MPI_Comm_size(comm, size);
  226. #endif
  227. }
  228. int starpu_mpi_comm_rank(MPI_Comm comm, int *rank)
  229. {
  230. if (_starpu_mpi_fake_world_rank != -1)
  231. {
  232. *rank = _starpu_mpi_fake_world_rank;
  233. return 0;
  234. }
  235. #ifdef STARPU_SIMGRID
  236. STARPU_MPI_ASSERT_MSG(comm == MPI_COMM_WORLD, "StarPU-SMPI only works with MPI_COMM_WORLD for now");
  237. *rank = _mpi_world_rank;
  238. return 0;
  239. #else
  240. return MPI_Comm_rank(comm, rank);
  241. #endif
  242. }
  243. int starpu_mpi_world_size(void)
  244. {
  245. int size;
  246. starpu_mpi_comm_size(MPI_COMM_WORLD, &size);
  247. return size;
  248. }
  249. int starpu_mpi_world_rank(void)
  250. {
  251. int rank;
  252. starpu_mpi_comm_rank(MPI_COMM_WORLD, &rank);
  253. return rank;
  254. }