starpu_mpi_insert_task.c 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2011, 2012 Centre National de la Recherche Scientifique
  4. * Copyright (C) 2011-2012 Université de Bordeaux 1
  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 <stdarg.h>
  18. #include <mpi.h>
  19. #include <starpu.h>
  20. #include <starpu_data.h>
  21. #include <common/utils.h>
  22. #include <common/uthash.h>
  23. #include <util/starpu_insert_task_utils.h>
  24. #include <datawizard/coherency.h>
  25. //#define STARPU_MPI_VERBOSE 1
  26. #include <starpu_mpi_private.h>
  27. /* Whether we are allowed to keep copies of remote data. Does not work
  28. * yet: the sender has to know whether the receiver has it, keeping it
  29. * in an array indexed by node numbers. */
  30. //#define MPI_CACHE 1
  31. #include <starpu_mpi_insert_task_cache.h>
  32. static void _starpu_mpi_tables_init()
  33. {
  34. if (sent_data == NULL) {
  35. int nb_nodes;
  36. int i;
  37. MPI_Comm_size(MPI_COMM_WORLD, &nb_nodes);
  38. _STARPU_MPI_DEBUG("Initialising htable for cache\n");
  39. sent_data = malloc(nb_nodes * sizeof(struct _starpu_data_entry *));
  40. for(i=0 ; i<nb_nodes ; i++) sent_data[i] = NULL;
  41. received_data = malloc(nb_nodes * sizeof(struct _starpu_data_entry *));
  42. for(i=0 ; i<nb_nodes ; i++) received_data[i] = NULL;
  43. }
  44. }
  45. void _starpu_data_deallocate(starpu_data_handle_t data_handle)
  46. {
  47. #ifdef STARPU_DEVEL
  48. #warning _starpu_data_deallocate not implemented yet
  49. #endif
  50. }
  51. int starpu_mpi_insert_task(MPI_Comm comm, struct starpu_codelet *codelet, ...)
  52. {
  53. int arg_type;
  54. va_list varg_list;
  55. int me, do_execute, xrank, nb_nodes;
  56. size_t *size_on_nodes;
  57. size_t arg_buffer_size = 0;
  58. char *arg_buffer;
  59. int dest=0, inconsistent_execute;
  60. _STARPU_MPI_LOG_IN();
  61. MPI_Comm_rank(comm, &me);
  62. MPI_Comm_size(comm, &nb_nodes);
  63. size_on_nodes = (size_t *)calloc(1, nb_nodes * sizeof(size_t));
  64. _starpu_mpi_tables_init();
  65. /* Get the number of buffers and the size of the arguments */
  66. va_start(varg_list, codelet);
  67. arg_buffer_size = _starpu_insert_task_get_arg_size(varg_list);
  68. va_start(varg_list, codelet);
  69. _starpu_codelet_pack_args(arg_buffer_size, &arg_buffer, varg_list);
  70. /* Find out whether we are to execute the data because we own the data to be written to. */
  71. inconsistent_execute = 0;
  72. do_execute = -1;
  73. xrank = -1;
  74. va_start(varg_list, codelet);
  75. while ((arg_type = va_arg(varg_list, int)) != 0) {
  76. if (arg_type==STARPU_EXECUTE_ON_NODE) {
  77. xrank = va_arg(varg_list, int);
  78. _STARPU_MPI_DEBUG("Executing on node %d\n", xrank);
  79. do_execute = 1;
  80. }
  81. else if (arg_type==STARPU_EXECUTE_ON_DATA) {
  82. starpu_data_handle_t data = va_arg(varg_list, starpu_data_handle_t);
  83. xrank = starpu_data_get_rank(data);
  84. _STARPU_MPI_DEBUG("Executing on data node %d\n", xrank);
  85. STARPU_ASSERT(xrank <= nb_nodes);
  86. do_execute = 1;
  87. }
  88. else if (arg_type==STARPU_R || arg_type==STARPU_W || arg_type==STARPU_RW || arg_type==STARPU_SCRATCH || arg_type==STARPU_REDUX) {
  89. starpu_data_handle_t data = va_arg(varg_list, starpu_data_handle_t);
  90. if (data && arg_type & STARPU_R) {
  91. int rank = starpu_data_get_rank(data);
  92. struct starpu_data_interface_ops *ops;
  93. ops = data->ops;
  94. size_on_nodes[rank] += ops->get_size(data);
  95. }
  96. if (arg_type & STARPU_W) {
  97. if (!data) {
  98. /* We don't have anything allocated for this.
  99. * The application knows we won't do anything
  100. * about this task */
  101. /* Yes, the app could actually not call
  102. * insert_task at all itself, this is just a
  103. * safeguard. */
  104. _STARPU_MPI_DEBUG("oh oh\n");
  105. _STARPU_MPI_LOG_OUT();
  106. free(size_on_nodes);
  107. return -EINVAL;
  108. }
  109. int mpi_rank = starpu_data_get_rank(data);
  110. if (mpi_rank == me) {
  111. if (do_execute == 0) {
  112. inconsistent_execute = 1;
  113. }
  114. else {
  115. do_execute = 1;
  116. }
  117. }
  118. else if (mpi_rank != -1) {
  119. if (do_execute == 1) {
  120. inconsistent_execute = 1;
  121. }
  122. else {
  123. do_execute = 0;
  124. dest = mpi_rank;
  125. /* That's the rank which needs the data to be sent to */
  126. }
  127. }
  128. else {
  129. _STARPU_ERROR("rank invalid\n");
  130. }
  131. }
  132. }
  133. else if (arg_type==STARPU_VALUE) {
  134. va_arg(varg_list, void *);
  135. va_arg(varg_list, size_t);
  136. }
  137. else if (arg_type==STARPU_CALLBACK) {
  138. va_arg(varg_list, void (*)(void *));
  139. }
  140. else if (arg_type==STARPU_CALLBACK_WITH_ARG) {
  141. va_arg(varg_list, void (*)(void *));
  142. va_arg(varg_list, void *);
  143. }
  144. else if (arg_type==STARPU_CALLBACK_ARG) {
  145. va_arg(varg_list, void *);
  146. }
  147. else if (arg_type==STARPU_PRIORITY) {
  148. va_arg(varg_list, int);
  149. }
  150. else if (arg_type==STARPU_EXECUTE_ON_NODE) {
  151. va_arg(varg_list, int);
  152. }
  153. else if (arg_type==STARPU_EXECUTE_ON_DATA) {
  154. va_arg(varg_list, int);
  155. }
  156. }
  157. va_end(varg_list);
  158. if (do_execute == -1) {
  159. int i;
  160. size_t max_size = 0;
  161. for(i=0 ; i<nb_nodes ; i++) {
  162. if (size_on_nodes[i] > max_size)
  163. {
  164. max_size = size_on_nodes[i];
  165. xrank = i;
  166. }
  167. }
  168. free(size_on_nodes);
  169. if (xrank != -1) {
  170. _STARPU_MPI_DEBUG("Node %d is having the most R data\n", xrank);
  171. do_execute = 1;
  172. }
  173. }
  174. STARPU_ASSERT(do_execute != -1 && "StarPU needs to see a W or a REDUX data which will tell it where to execute the task");
  175. if (inconsistent_execute == 1) {
  176. if (xrank == -1) {
  177. _STARPU_MPI_DEBUG("Different tasks are owning W data. Needs to specify which one is to execute the codelet, using STARPU_EXECUTE_ON_NODE or STARPU_EXECUTE_ON_DATA\n");
  178. free(size_on_nodes);
  179. return -EINVAL;
  180. }
  181. else {
  182. do_execute = (me == xrank);
  183. dest = xrank;
  184. }
  185. }
  186. else if (xrank != -1) {
  187. do_execute = (me == xrank);
  188. dest = xrank;
  189. }
  190. /* Send and receive data as requested */
  191. va_start(varg_list, codelet);
  192. while ((arg_type = va_arg(varg_list, int)) != 0) {
  193. if (arg_type==STARPU_R || arg_type==STARPU_W || arg_type==STARPU_RW || arg_type == STARPU_SCRATCH) {
  194. starpu_data_handle_t data = va_arg(varg_list, starpu_data_handle_t);
  195. if (data && arg_type & STARPU_R) {
  196. int mpi_rank = starpu_data_get_rank(data);
  197. int mpi_tag = starpu_data_get_tag(data);
  198. STARPU_ASSERT(mpi_tag >= 0 && "StarPU needs to be told the MPI rank of this data, using starpu_data_set_rank");
  199. /* The task needs to read this data */
  200. if (do_execute && mpi_rank != me && mpi_rank != -1) {
  201. /* I will have to execute but I don't have the data, receive */
  202. #ifdef MPI_CACHE
  203. struct _starpu_data_entry *already_received;
  204. HASH_FIND_PTR(received_data[mpi_rank], &data, already_received);
  205. if (already_received == NULL) {
  206. struct _starpu_data_entry *entry = (struct _starpu_data_entry *)malloc(sizeof(*entry));
  207. entry->data = data;
  208. HASH_ADD_PTR(received_data[mpi_rank], data, entry);
  209. }
  210. else {
  211. _STARPU_MPI_DEBUG("Do not receive data %p from node %d as it is already available\n", data, mpi_rank);
  212. }
  213. if (!already_received)
  214. #endif
  215. {
  216. _STARPU_MPI_DEBUG("Receive data %p from %d\n", data, mpi_rank);
  217. starpu_mpi_irecv_detached(data, mpi_rank, mpi_tag, comm, NULL, NULL);
  218. }
  219. }
  220. if (!do_execute && mpi_rank == me) {
  221. /* Somebody else will execute it, and I have the data, send it. */
  222. #ifdef MPI_CACHE
  223. struct _starpu_data_entry *already_sent;
  224. HASH_FIND_PTR(sent_data[mpi_rank], &data, already_sent);
  225. if (already_sent == NULL) {
  226. struct _starpu_data_entry *entry = (struct _starpu_data_entry *)malloc(sizeof(*entry));
  227. entry->data = data;
  228. HASH_ADD_PTR(sent_data[dest], data, entry);
  229. }
  230. else {
  231. _STARPU_MPI_DEBUG("Do not send data %p to node %d as it has already been sent\n", data, dest);
  232. }
  233. if (!already_sent)
  234. #endif
  235. {
  236. _STARPU_MPI_DEBUG("Send data %p to %d\n", data, dest);
  237. starpu_mpi_isend_detached(data, dest, mpi_tag, comm, NULL, NULL);
  238. }
  239. }
  240. }
  241. }
  242. else if (arg_type==STARPU_VALUE) {
  243. va_arg(varg_list, void *);
  244. va_arg(varg_list, size_t);
  245. }
  246. else if (arg_type==STARPU_CALLBACK) {
  247. va_arg(varg_list, void (*)(void *));
  248. }
  249. else if (arg_type==STARPU_CALLBACK_WITH_ARG) {
  250. va_arg(varg_list, void (*)(void *));
  251. va_arg(varg_list, void *);
  252. }
  253. else if (arg_type==STARPU_CALLBACK_ARG) {
  254. va_arg(varg_list, void *);
  255. }
  256. else if (arg_type==STARPU_PRIORITY) {
  257. va_arg(varg_list, int);
  258. }
  259. else if (arg_type==STARPU_EXECUTE_ON_NODE) {
  260. va_arg(varg_list, int);
  261. }
  262. else if (arg_type==STARPU_EXECUTE_ON_DATA) {
  263. va_arg(varg_list, starpu_data_handle_t);
  264. }
  265. }
  266. va_end(varg_list);
  267. if (do_execute) {
  268. _STARPU_MPI_DEBUG("Execution of the codelet %p (%s)\n", codelet, codelet->name);
  269. va_start(varg_list, codelet);
  270. struct starpu_task *task = starpu_task_create();
  271. int ret = _starpu_insert_task_create_and_submit(arg_buffer, arg_buffer_size, codelet, &task, varg_list);
  272. _STARPU_MPI_DEBUG("ret: %d\n", ret);
  273. STARPU_ASSERT(ret==0);
  274. }
  275. if (inconsistent_execute) {
  276. va_start(varg_list, codelet);
  277. while ((arg_type = va_arg(varg_list, int)) != 0) {
  278. if (arg_type==STARPU_R || arg_type==STARPU_W || arg_type==STARPU_RW || arg_type == STARPU_SCRATCH) {
  279. starpu_data_handle_t data = va_arg(varg_list, starpu_data_handle_t);
  280. if (arg_type & STARPU_W) {
  281. int mpi_rank = starpu_data_get_rank(data);
  282. int mpi_tag = starpu_data_get_tag(data);
  283. STARPU_ASSERT(mpi_tag >= 0 && "StarPU needs to be told the MPI rank of this data, using starpu_data_set_rank");
  284. if (mpi_rank == me) {
  285. if (xrank != -1 && me != xrank) {
  286. _STARPU_MPI_DEBUG("Receive data %p back from the task %d which executed the codelet ...\n", data, dest);
  287. starpu_mpi_irecv_detached(data, dest, mpi_tag, comm, NULL, NULL);
  288. }
  289. }
  290. else if (do_execute) {
  291. _STARPU_MPI_DEBUG("Send data %p back to its owner %d...\n", data, mpi_rank);
  292. starpu_mpi_isend_detached(data, mpi_rank, mpi_tag, comm, NULL, NULL);
  293. }
  294. }
  295. }
  296. else if (arg_type==STARPU_VALUE) {
  297. va_arg(varg_list, void *);
  298. va_arg(varg_list, size_t);
  299. }
  300. else if (arg_type==STARPU_CALLBACK) {
  301. va_arg(varg_list, void (*)(void *));
  302. }
  303. else if (arg_type==STARPU_CALLBACK_WITH_ARG) {
  304. va_arg(varg_list, void (*)(void *));
  305. va_arg(varg_list, void *);
  306. }
  307. else if (arg_type==STARPU_CALLBACK_ARG) {
  308. va_arg(varg_list, void *);
  309. }
  310. else if (arg_type==STARPU_PRIORITY) {
  311. va_arg(varg_list, int);
  312. }
  313. else if (arg_type==STARPU_EXECUTE_ON_NODE) {
  314. va_arg(varg_list, int);
  315. }
  316. else if (arg_type==STARPU_EXECUTE_ON_DATA) {
  317. va_arg(varg_list, starpu_data_handle_t);
  318. }
  319. }
  320. va_end(varg_list);
  321. }
  322. va_start(varg_list, codelet);
  323. while ((arg_type = va_arg(varg_list, int)) != 0) {
  324. if (arg_type==STARPU_R || arg_type==STARPU_W || arg_type==STARPU_RW || arg_type == STARPU_SCRATCH) {
  325. starpu_data_handle_t data = va_arg(varg_list, starpu_data_handle_t);
  326. #ifdef MPI_CACHE
  327. if (arg_type & STARPU_W) {
  328. if (do_execute) {
  329. /* Note that all copies I've sent to neighbours are now invalid */
  330. int n, size;
  331. MPI_Comm_size(comm, &size);
  332. for(n=0 ; n<size ; n++) {
  333. struct _starpu_data_entry *already_sent;
  334. HASH_FIND_PTR(sent_data[n], &data, already_sent);
  335. if (already_sent) {
  336. _STARPU_MPI_DEBUG("Posting request to clear send cache for data %p\n", data);
  337. _starpu_mpi_clear_cache_request(data, n, _STARPU_MPI_CLEAR_SENT_DATA);
  338. }
  339. }
  340. }
  341. else {
  342. int mpi_rank = starpu_data_get_rank(data);
  343. struct _starpu_data_entry *already_received;
  344. HASH_FIND_PTR(received_data[mpi_rank], &data, already_received);
  345. if (already_received) {
  346. /* Somebody else will write to the data, so discard our cached copy if any */
  347. /* TODO: starpu_mpi could just remember itself. */
  348. _STARPU_MPI_DEBUG("Posting request to clear receive cache for data %p\n", data);
  349. _starpu_mpi_clear_cache_request(data, mpi_rank, _STARPU_MPI_CLEAR_RECEIVED_DATA);
  350. _starpu_data_deallocate(data);
  351. }
  352. }
  353. }
  354. #else
  355. /* We allocated a temporary buffer for the received data, now drop it */
  356. if ((arg_type & STARPU_R) && do_execute) {
  357. int mpi_rank = starpu_data_get_rank(data);
  358. if (mpi_rank != me && mpi_rank != -1) {
  359. _starpu_data_deallocate(data);
  360. }
  361. }
  362. #endif
  363. }
  364. else if (arg_type==STARPU_VALUE) {
  365. va_arg(varg_list, void *);
  366. va_arg(varg_list, size_t);
  367. }
  368. else if (arg_type==STARPU_CALLBACK) {
  369. va_arg(varg_list, void (*)(void *));
  370. }
  371. else if (arg_type==STARPU_CALLBACK_WITH_ARG) {
  372. va_arg(varg_list, void (*)(void *));
  373. va_arg(varg_list, void *);
  374. }
  375. else if (arg_type==STARPU_CALLBACK_ARG) {
  376. va_arg(varg_list, void *);
  377. }
  378. else if (arg_type==STARPU_PRIORITY) {
  379. va_arg(varg_list, int);
  380. }
  381. else if (arg_type==STARPU_EXECUTE_ON_NODE) {
  382. va_arg(varg_list, int);
  383. }
  384. else if (arg_type==STARPU_EXECUTE_ON_DATA) {
  385. va_arg(varg_list, starpu_data_handle_t);
  386. }
  387. }
  388. va_end(varg_list);
  389. _STARPU_MPI_LOG_OUT();
  390. return 0;
  391. }
  392. void starpu_mpi_get_data_on_node_detached(MPI_Comm comm, starpu_data_handle_t data_handle, int node, void (*callback)(void*), void *arg)
  393. {
  394. int me, rank, tag;
  395. rank = starpu_data_get_rank(data_handle);
  396. tag = starpu_data_get_tag(data_handle);
  397. MPI_Comm_rank(comm, &me);
  398. if (node == rank) return;
  399. if (me == node)
  400. {
  401. starpu_mpi_irecv_detached(data_handle, rank, tag, comm, callback, arg);
  402. }
  403. else if (me == rank)
  404. {
  405. starpu_mpi_isend_detached(data_handle, node, tag, comm, NULL, NULL);
  406. }
  407. }
  408. void starpu_mpi_get_data_on_node(MPI_Comm comm, starpu_data_handle_t data_handle, int node)
  409. {
  410. int me, rank, tag;
  411. rank = starpu_data_get_rank(data_handle);
  412. tag = starpu_data_get_tag(data_handle);
  413. MPI_Comm_rank(comm, &me);
  414. if (node == rank) return;
  415. if (me == node)
  416. {
  417. MPI_Status status;
  418. starpu_mpi_recv(data_handle, rank, tag, comm, &status);
  419. }
  420. else if (me == rank)
  421. {
  422. starpu_mpi_send(data_handle, node, tag, comm);
  423. }
  424. }
  425. void starpu_mpi_redux_data(MPI_Comm comm, starpu_data_handle_t data_handle)
  426. {
  427. int me, rank, tag, nb_nodes;
  428. rank = starpu_data_get_rank(data_handle);
  429. tag = starpu_data_get_tag(data_handle);
  430. MPI_Comm_rank(comm, &me);
  431. MPI_Comm_size(comm, &nb_nodes);
  432. _STARPU_MPI_DEBUG("Doing reduction for data %p on node %d with %d nodes ...\n", data_handle, rank, nb_nodes);
  433. // need to count how many nodes have the data in redux mode
  434. if (me == rank) {
  435. int i;
  436. for(i=0 ; i<nb_nodes ; i++) {
  437. if (i != rank) {
  438. starpu_data_handle_t new_handle;
  439. starpu_data_register_same(&new_handle, data_handle);
  440. _STARPU_MPI_DEBUG("Receiving redux handle from %d in %p ...\n", i, new_handle);
  441. starpu_mpi_irecv_detached(new_handle, i, tag, comm, NULL, NULL);
  442. starpu_insert_task(data_handle->redux_cl,
  443. STARPU_RW, data_handle,
  444. STARPU_R, new_handle,
  445. 0);
  446. }
  447. }
  448. }
  449. else {
  450. _STARPU_MPI_DEBUG("Sending redux handle to %d ...\n", rank);
  451. starpu_mpi_isend_detached(data_handle, rank, tag, comm, NULL, NULL);
  452. }
  453. }