starpu_mpi_init.c 7.2 KB

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