starpu_mpi.c 56 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2009, 2010-2013 Université de Bordeaux 1
  4. * Copyright (C) 2010, 2011, 2012, 2013 Centre National de la Recherche Scientifique
  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 <stdlib.h>
  18. #include <starpu_mpi.h>
  19. #include <starpu_mpi_datatype.h>
  20. #include <starpu_mpi_private.h>
  21. #include <starpu_profiling.h>
  22. #include <starpu_mpi_stats.h>
  23. #include <starpu_mpi_task_insert.h>
  24. #include <common/config.h>
  25. #include <common/thread.h>
  26. static void _starpu_mpi_add_sync_point_in_fxt(void);
  27. static void _starpu_mpi_submit_new_mpi_request(void *arg);
  28. static void _starpu_mpi_handle_request_termination(struct _starpu_mpi_req *req);
  29. #ifdef STARPU_VERBOSE
  30. static char *_starpu_mpi_request_type(enum _starpu_mpi_request_type request_type);
  31. #endif
  32. static struct _starpu_mpi_req *_starpu_mpi_isend_common(starpu_data_handle_t data_handle,
  33. int dest, int mpi_tag, MPI_Comm comm,
  34. unsigned detached, void (*callback)(void *), void *arg,
  35. int sequential_consistency);
  36. static struct _starpu_mpi_req *_starpu_mpi_irecv_common(starpu_data_handle_t data_handle,
  37. int source, int mpi_tag, MPI_Comm comm,
  38. unsigned detached, void (*callback)(void *), void *arg,
  39. int sequential_consistency, int is_internal_req,
  40. ssize_t count);
  41. static void _starpu_mpi_handle_detached_request(struct _starpu_mpi_req *req);
  42. /* The list of requests that have been newly submitted by the application */
  43. static struct _starpu_mpi_req_list *new_requests;
  44. /* The list of detached requests that have already been submitted to MPI */
  45. static struct _starpu_mpi_req_list *detached_requests;
  46. static starpu_pthread_mutex_t detached_requests_mutex;
  47. /* Condition to wake up progression thread */
  48. static starpu_pthread_cond_t cond_progression;
  49. /* Condition to wake up waiting for all current MPI requests to finish */
  50. static starpu_pthread_cond_t cond_finished;
  51. static starpu_pthread_mutex_t mutex;
  52. static starpu_pthread_t progress_thread;
  53. static int running = 0;
  54. /* Count requests posted by the application and not yet submitted to MPI, i.e pushed into the new_requests list */
  55. static starpu_pthread_mutex_t mutex_posted_requests;
  56. static int posted_requests = 0, newer_requests, barrier_running = 0;
  57. #define _STARPU_MPI_INC_POSTED_REQUESTS(value) { STARPU_PTHREAD_MUTEX_LOCK(&mutex_posted_requests); posted_requests += value; STARPU_PTHREAD_MUTEX_UNLOCK(&mutex_posted_requests); }
  58. LIST_TYPE(_starpu_mpi_copy_handle,
  59. starpu_data_handle_t handle;
  60. struct _starpu_mpi_envelope *env;
  61. struct _starpu_mpi_req *req;
  62. void *buffer;
  63. int mpi_tag;
  64. int source;
  65. int req_ready;
  66. starpu_pthread_mutex_t req_mutex;
  67. starpu_pthread_cond_t req_cond;
  68. );
  69. struct _starpu_mpi_copy_handle_hashlist
  70. {
  71. struct _starpu_mpi_copy_handle_list *list;
  72. UT_hash_handle hh;
  73. int mpi_tag;
  74. };
  75. /********************************************************/
  76. /* */
  77. /* Hashmap's requests functionalities */
  78. /* */
  79. /********************************************************/
  80. /** stores application requests for which data have not been received yet */
  81. static struct _starpu_mpi_req **_starpu_mpi_app_req_hashmap = NULL;
  82. static int _starpu_mpi_app_req_hashmap_count = 0;
  83. /** stores data which have been received by MPI but have not been requested by the application */
  84. static struct _starpu_mpi_copy_handle_hashlist **_starpu_mpi_copy_handle_hashmap = NULL;
  85. static int _starpu_mpi_copy_handle_hashmap_count = 0;
  86. static struct _starpu_mpi_req* find_app_req(int mpi_tag, int source)
  87. {
  88. struct _starpu_mpi_req* req;
  89. HASH_FIND_INT(_starpu_mpi_app_req_hashmap[source], &mpi_tag, req);
  90. return req;
  91. }
  92. static void add_app_req(struct _starpu_mpi_req *req)
  93. {
  94. struct _starpu_mpi_req *test_req;
  95. test_req = find_app_req(req->mpi_tag, req->srcdst);
  96. if (test_req == NULL)
  97. {
  98. HASH_ADD_INT(_starpu_mpi_app_req_hashmap[req->srcdst], mpi_tag, req);
  99. _starpu_mpi_app_req_hashmap_count ++;
  100. _STARPU_MPI_DEBUG(3, "Adding request %p with tag %d in the application request hashmap[%d]\n", req, req->mpi_tag, req->srcdst);
  101. }
  102. else
  103. {
  104. _STARPU_MPI_DEBUG(3, "[Error] request %p with tag %d already in the application request hashmap[%d]\n", req, req->mpi_tag, req->srcdst);
  105. int seq_const = starpu_data_get_sequential_consistency_flag(req->data_handle);
  106. if (seq_const && req->sequential_consistency)
  107. {
  108. STARPU_ASSERT_MSG(!test_req, "[Error] request %p with tag %d wanted to be added to the application request hashmap[%d], while another request %p with the same tag is already in it. \n Sequential consistency is activated : this is not supported by StarPU.", req, req->mpi_tag, req->srcdst, test_req);
  109. }
  110. else
  111. {
  112. STARPU_ASSERT_MSG(!test_req, "[Error] request %p with tag %d wanted to be added to the application request hashmap[%d], while another request %p with the same tag is already in it. \n Sequential consistency isn't activated for this handle : you should want to add dependencies between requests for which the sequential consistency is deactivated.", req, req->mpi_tag, req->srcdst, test_req);
  113. }
  114. }
  115. }
  116. static void delete_app_req(struct _starpu_mpi_req *req)
  117. {
  118. struct _starpu_mpi_req *test_req;
  119. test_req = find_app_req(req->mpi_tag, req->srcdst);
  120. if (test_req != NULL)
  121. {
  122. HASH_DEL(_starpu_mpi_app_req_hashmap[req->srcdst], req);
  123. _starpu_mpi_app_req_hashmap_count --;
  124. _STARPU_MPI_DEBUG(3, "Deleting application request %p with tag %d from the application request hashmap[%d]\n", req, req->mpi_tag, req->srcdst);
  125. }
  126. else
  127. {
  128. _STARPU_MPI_DEBUG(3, "[Warning] request %p with tag %d is NOT in the application request hashmap[%d]\n", req, req->mpi_tag, req->srcdst);
  129. }
  130. }
  131. #ifdef STARPU_VERBOSE
  132. static void _starpu_mpi_copy_handle_display_hash(int source, int tag)
  133. {
  134. struct _starpu_mpi_copy_handle_hashlist *hashlist;
  135. HASH_FIND_INT(_starpu_mpi_copy_handle_hashmap[source], &tag, hashlist);
  136. if (hashlist == NULL)
  137. {
  138. _STARPU_MPI_DEBUG(60, "Hashlist for source %d and tag %d does not exist\n", source, tag);
  139. }
  140. else if (_starpu_mpi_copy_handle_list_empty(hashlist->list))
  141. {
  142. _STARPU_MPI_DEBUG(60, "Hashlist for source %d and tag %d is empty\n", source, tag);
  143. }
  144. else
  145. {
  146. struct _starpu_mpi_copy_handle *cur;
  147. for (cur = _starpu_mpi_copy_handle_list_begin(hashlist->list) ;
  148. cur != _starpu_mpi_copy_handle_list_end(hashlist->list);
  149. cur = _starpu_mpi_copy_handle_list_next(cur))
  150. {
  151. _STARPU_MPI_DEBUG(60, "Element for source %d and tag %d: %p\n", source, tag, cur);
  152. }
  153. }
  154. }
  155. #endif
  156. static struct _starpu_mpi_copy_handle *pop_chandle(int mpi_tag, int source, int delete)
  157. {
  158. struct _starpu_mpi_copy_handle_hashlist *hashlist;
  159. struct _starpu_mpi_copy_handle *chandle;
  160. _STARPU_MPI_DEBUG(60, "Looking for chandle with tag %d in the hashmap[%d]\n", mpi_tag, source);
  161. HASH_FIND_INT(_starpu_mpi_copy_handle_hashmap[source], &mpi_tag, hashlist);
  162. if (hashlist == NULL)
  163. {
  164. chandle = NULL;
  165. }
  166. else
  167. {
  168. if (_starpu_mpi_copy_handle_list_empty(hashlist->list))
  169. {
  170. chandle = NULL;
  171. }
  172. else
  173. {
  174. if (delete == 1)
  175. {
  176. chandle = _starpu_mpi_copy_handle_list_pop_front(hashlist->list);
  177. }
  178. else
  179. {
  180. chandle = _starpu_mpi_copy_handle_list_front(hashlist->list);
  181. }
  182. }
  183. }
  184. _STARPU_MPI_DEBUG(60, "Found chandle %p with tag %d in the hashmap[%d]\n", chandle, mpi_tag, source);
  185. return chandle;
  186. }
  187. static struct _starpu_mpi_copy_handle *find_chandle(int mpi_tag, int source)
  188. {
  189. return pop_chandle(mpi_tag, source, 0);
  190. }
  191. static void add_chandle(struct _starpu_mpi_copy_handle *chandle)
  192. {
  193. _STARPU_MPI_DEBUG(60, "Trying to add chandle %p with tag %d in the hashmap[%d]\n", chandle, chandle->mpi_tag, chandle->source);
  194. struct _starpu_mpi_copy_handle_hashlist *hashlist;
  195. HASH_FIND_INT(_starpu_mpi_copy_handle_hashmap[chandle->source], &chandle->mpi_tag, hashlist);
  196. if (hashlist == NULL)
  197. {
  198. hashlist = malloc(sizeof(struct _starpu_mpi_copy_handle_hashlist));
  199. hashlist->list = _starpu_mpi_copy_handle_list_new();
  200. hashlist->mpi_tag = chandle->mpi_tag;
  201. HASH_ADD_INT(_starpu_mpi_copy_handle_hashmap[chandle->source], mpi_tag, hashlist);
  202. }
  203. _starpu_mpi_copy_handle_list_push_back(hashlist->list, chandle);
  204. _starpu_mpi_copy_handle_hashmap_count ++;
  205. #ifdef STARPU_VERBOSE
  206. _starpu_mpi_copy_handle_display_hash(chandle->source, chandle->mpi_tag);
  207. #endif
  208. }
  209. static void delete_chandle(struct _starpu_mpi_copy_handle *chandle)
  210. {
  211. _STARPU_MPI_DEBUG(60, "Trying to delete chandle %p with tag %d in the hashmap[%d]\n", chandle, chandle->mpi_tag, chandle->source);
  212. struct _starpu_mpi_copy_handle *found = pop_chandle(chandle->mpi_tag, chandle->source, 1);
  213. STARPU_ASSERT_MSG(found == chandle,
  214. "Error delete_chandle : chandle %p with tag %d is NOT in the hashmap[%d]\n", chandle, chandle->mpi_tag, chandle->source);
  215. _starpu_mpi_copy_handle_hashmap_count --;
  216. #ifdef STARPU_VERBOSE
  217. _starpu_mpi_copy_handle_display_hash(chandle->source, chandle->mpi_tag);
  218. #endif
  219. }
  220. static void _starpu_mpi_request_init(struct _starpu_mpi_req *req)
  221. {
  222. /* Initialize the request structure */
  223. req->data_handle = NULL;
  224. req->datatype = 0;
  225. req->ptr = NULL;
  226. req->count = -1;
  227. req->user_datatype = -1;
  228. req->srcdst = -1;
  229. req->mpi_tag = -1;
  230. req->comm = 0;
  231. req->func = NULL;
  232. req->status = NULL;
  233. req->request = 0;
  234. req->flag = NULL;
  235. req->ret = -1;
  236. STARPU_PTHREAD_MUTEX_INIT(&req->req_mutex, NULL);
  237. STARPU_PTHREAD_COND_INIT(&req->req_cond, NULL);
  238. STARPU_PTHREAD_MUTEX_INIT(&req->posted_mutex, NULL);
  239. STARPU_PTHREAD_COND_INIT(&req->posted_cond, NULL);
  240. req->request_type = UNKNOWN_REQ;
  241. req->submitted = 0;
  242. req->completed = 0;
  243. req->posted = 0;
  244. req->other_request = NULL;
  245. req->detached = -1;
  246. req->callback = NULL;
  247. req->callback_arg = NULL;
  248. req->size_req = 0;
  249. req->internal_req = NULL;
  250. req->is_internal_req = 0;
  251. req->envelope = NULL;
  252. req->sequential_consistency = 1;
  253. }
  254. /********************************************************/
  255. /* */
  256. /* Send/Receive functionalities */
  257. /* */
  258. /********************************************************/
  259. static struct _starpu_mpi_req *_starpu_mpi_isend_irecv_common(starpu_data_handle_t data_handle,
  260. int srcdst, int mpi_tag, MPI_Comm comm,
  261. unsigned detached, void (*callback)(void *), void *arg,
  262. enum _starpu_mpi_request_type request_type, void (*func)(struct _starpu_mpi_req *),
  263. enum starpu_data_access_mode mode,
  264. int sequential_consistency,
  265. int is_internal_req,
  266. ssize_t count)
  267. {
  268. _STARPU_MPI_LOG_IN();
  269. struct _starpu_mpi_req *req = malloc(sizeof(struct _starpu_mpi_req));
  270. STARPU_ASSERT_MSG(req, "Invalid request");
  271. _STARPU_MPI_INC_POSTED_REQUESTS(1);
  272. /* Initialize the request structure */
  273. _starpu_mpi_request_init(req);
  274. req->request_type = request_type;
  275. req->data_handle = data_handle;
  276. req->srcdst = srcdst;
  277. req->mpi_tag = mpi_tag;
  278. req->comm = comm;
  279. req->detached = detached;
  280. req->callback = callback;
  281. req->callback_arg = arg;
  282. req->func = func;
  283. req->sequential_consistency = sequential_consistency;
  284. req->is_internal_req = is_internal_req;
  285. req->count = count;
  286. /* Asynchronously request StarPU to fetch the data in main memory: when
  287. * it is available in main memory, _starpu_mpi_submit_new_mpi_request(req) is called and
  288. * the request is actually submitted */
  289. starpu_data_acquire_cb_sequential_consistency(data_handle, mode, _starpu_mpi_submit_new_mpi_request, (void *)req, sequential_consistency);
  290. _STARPU_MPI_LOG_OUT();
  291. return req;
  292. }
  293. /********************************************************/
  294. /* */
  295. /* Send functionalities */
  296. /* */
  297. /********************************************************/
  298. static void _starpu_mpi_isend_data_func(struct _starpu_mpi_req *req)
  299. {
  300. _STARPU_MPI_LOG_IN();
  301. STARPU_ASSERT_MSG(req->ptr, "Pointer containing data to send is invalid");
  302. _STARPU_MPI_DEBUG(2, "post MPI isend request %p type %s tag %d src %d data %p datasize %ld ptr %p datatype '%s' count %d user_datatype %d \n", req, _starpu_mpi_request_type(req->request_type), req->mpi_tag, req->srcdst, req->data_handle, starpu_data_get_size(req->data_handle), req->ptr, _starpu_mpi_datatype(req->datatype), (int)req->count, req->user_datatype);
  303. _starpu_mpi_comm_amounts_inc(req->comm, req->srcdst, req->datatype, req->count);
  304. TRACE_MPI_ISEND_SUBMIT_BEGIN(req->srcdst, req->mpi_tag, 0);
  305. req->ret = MPI_Isend(req->ptr, req->count, req->datatype, req->srcdst, _starpu_mpi_tag, req->comm, &req->request);
  306. STARPU_ASSERT_MSG(req->ret == MPI_SUCCESS, "MPI_Isend returning %d", req->ret);
  307. TRACE_MPI_ISEND_SUBMIT_END(req->srcdst, req->mpi_tag, 0);
  308. /* somebody is perhaps waiting for the MPI request to be posted */
  309. STARPU_PTHREAD_MUTEX_LOCK(&req->req_mutex);
  310. req->submitted = 1;
  311. STARPU_PTHREAD_COND_BROADCAST(&req->req_cond);
  312. STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
  313. _starpu_mpi_handle_detached_request(req);
  314. _STARPU_MPI_LOG_OUT();
  315. }
  316. static void _starpu_mpi_isend_size_func(struct _starpu_mpi_req *req)
  317. {
  318. _starpu_mpi_handle_allocate_datatype(req->data_handle, &req->datatype, &req->user_datatype);
  319. req->envelope = calloc(1,sizeof(struct _starpu_mpi_envelope));
  320. req->envelope->mpi_tag = req->mpi_tag;
  321. if (req->user_datatype == 0)
  322. {
  323. int size;
  324. req->count = 1;
  325. req->ptr = starpu_data_get_local_ptr(req->data_handle);
  326. MPI_Type_size(req->datatype, &size);
  327. req->envelope->size = (ssize_t)req->count * size;
  328. _STARPU_MPI_DEBUG(1, "Post MPI isend count (%ld) datatype_size %ld request to %d with tag %d\n",req->count,starpu_data_get_size(req->data_handle),req->srcdst, _starpu_mpi_tag);
  329. MPI_Isend(req->envelope, sizeof(struct _starpu_mpi_envelope), MPI_BYTE, req->srcdst, _starpu_mpi_tag, req->comm, &req->size_req);
  330. }
  331. else
  332. {
  333. int ret;
  334. // Do not pack the data, just try to find out the size
  335. starpu_data_pack(req->data_handle, NULL, &(req->envelope->size));
  336. if (req->envelope->size != -1)
  337. {
  338. // We already know the size of the data, let's send it to overlap with the packing of the data
  339. _STARPU_MPI_DEBUG(1, "Sending size %ld (%ld %s) with tag %d to node %d (first call to pack)\n", req->envelope->size, sizeof(req->count), _starpu_mpi_datatype(MPI_BYTE), _starpu_mpi_tag, req->srcdst);
  340. req->count = req->envelope->size;
  341. ret = MPI_Isend(req->envelope, sizeof(struct _starpu_mpi_envelope), MPI_BYTE, req->srcdst, _starpu_mpi_tag, req->comm, &req->size_req);
  342. STARPU_ASSERT_MSG(ret == MPI_SUCCESS, "when sending size, MPI_Isend returning %d", ret);
  343. }
  344. // Pack the data
  345. starpu_data_pack(req->data_handle, &req->ptr, &req->count);
  346. if (req->envelope->size == -1)
  347. {
  348. // We know the size now, let's send it
  349. _STARPU_MPI_DEBUG(1, "Sending size %ld (%ld %s) with tag %d to node %d (second call to pack)\n", req->envelope->size, sizeof(req->count), _starpu_mpi_datatype(MPI_BYTE), _starpu_mpi_tag, req->srcdst);
  350. ret = MPI_Isend(req->envelope, sizeof(struct _starpu_mpi_envelope), MPI_BYTE, req->srcdst, _starpu_mpi_tag, req->comm, &req->size_req);
  351. STARPU_ASSERT_MSG(ret == MPI_SUCCESS, "when sending size, MPI_Isend returning %d", ret);
  352. }
  353. else
  354. {
  355. // We check the size returned with the 2 calls to pack is the same
  356. STARPU_ASSERT_MSG(req->count == req->envelope->size, "Calls to pack_data returned different sizes %ld != %ld", req->count, req->envelope->size);
  357. }
  358. // We can send the data now
  359. }
  360. _starpu_mpi_isend_data_func(req);
  361. }
  362. static struct _starpu_mpi_req *_starpu_mpi_isend_common(starpu_data_handle_t data_handle,
  363. int dest, int mpi_tag, MPI_Comm comm,
  364. unsigned detached, void (*callback)(void *), void *arg,
  365. int sequential_consistency)
  366. {
  367. return _starpu_mpi_isend_irecv_common(data_handle, dest, mpi_tag, comm, detached, callback, arg, SEND_REQ, _starpu_mpi_isend_size_func, STARPU_R, sequential_consistency, 0, 0);
  368. }
  369. int starpu_mpi_isend(starpu_data_handle_t data_handle, starpu_mpi_req *public_req, int dest, int mpi_tag, MPI_Comm comm)
  370. {
  371. _STARPU_MPI_LOG_IN();
  372. STARPU_ASSERT_MSG(public_req, "starpu_mpi_isend needs a valid starpu_mpi_req");
  373. struct _starpu_mpi_req *req;
  374. req = _starpu_mpi_isend_common(data_handle, dest, mpi_tag, comm, 0, NULL, NULL, 1);
  375. STARPU_ASSERT_MSG(req, "Invalid return for _starpu_mpi_isend_common");
  376. *public_req = req;
  377. _STARPU_MPI_LOG_OUT();
  378. return 0;
  379. }
  380. int starpu_mpi_isend_detached(starpu_data_handle_t data_handle,
  381. int dest, int mpi_tag, MPI_Comm comm, void (*callback)(void *), void *arg)
  382. {
  383. _STARPU_MPI_LOG_IN();
  384. _starpu_mpi_isend_common(data_handle, dest, mpi_tag, comm, 1, callback, arg, 1);
  385. _STARPU_MPI_LOG_OUT();
  386. return 0;
  387. }
  388. int starpu_mpi_send(starpu_data_handle_t data_handle, int dest, int mpi_tag, MPI_Comm comm)
  389. {
  390. starpu_mpi_req req;
  391. MPI_Status status;
  392. _STARPU_MPI_LOG_IN();
  393. memset(&status, 0, sizeof(MPI_Status));
  394. starpu_mpi_isend(data_handle, &req, dest, mpi_tag, comm);
  395. starpu_mpi_wait(&req, &status);
  396. _STARPU_MPI_LOG_OUT();
  397. return 0;
  398. }
  399. /********************************************************/
  400. /* */
  401. /* receive functionalities */
  402. /* */
  403. /********************************************************/
  404. static void _starpu_mpi_irecv_data_func(struct _starpu_mpi_req *req)
  405. {
  406. _STARPU_MPI_LOG_IN();
  407. STARPU_ASSERT_MSG(req->ptr, "Invalid pointer to receive data");
  408. _STARPU_MPI_DEBUG(20, "post MPI irecv request %p type %s tag %d src %d data %p ptr %p datatype '%s' count %d user_datatype %d \n", req, _starpu_mpi_request_type(req->request_type), req->mpi_tag, req->srcdst, req->data_handle, req->ptr, _starpu_mpi_datatype(req->datatype), (int)req->count, req->user_datatype);
  409. TRACE_MPI_IRECV_SUBMIT_BEGIN(req->srcdst, req->mpi_tag);
  410. req->ret = MPI_Irecv(req->ptr, req->count, req->datatype, req->srcdst, _starpu_mpi_tag, req->comm, &req->request);
  411. STARPU_ASSERT_MSG(req->ret == MPI_SUCCESS, "MPI_IRecv returning %d", req->ret);
  412. TRACE_MPI_IRECV_SUBMIT_END(req->srcdst, req->mpi_tag);
  413. /* somebody is perhaps waiting for the MPI request to be posted */
  414. STARPU_PTHREAD_MUTEX_LOCK(&req->req_mutex);
  415. req->submitted = 1;
  416. STARPU_PTHREAD_COND_BROADCAST(&req->req_cond);
  417. STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
  418. _starpu_mpi_handle_detached_request(req);
  419. _STARPU_MPI_LOG_OUT();
  420. }
  421. static struct _starpu_mpi_req *_starpu_mpi_irecv_common(starpu_data_handle_t data_handle, int source, int mpi_tag, MPI_Comm comm, unsigned detached, void (*callback)(void *), void *arg, int sequential_consistency, int is_internal_req, ssize_t count)
  422. {
  423. return _starpu_mpi_isend_irecv_common(data_handle, source, mpi_tag, comm, detached, callback, arg, RECV_REQ, _starpu_mpi_irecv_data_func, STARPU_W, sequential_consistency, is_internal_req, count);
  424. }
  425. int starpu_mpi_irecv(starpu_data_handle_t data_handle, starpu_mpi_req *public_req, int source, int mpi_tag, MPI_Comm comm)
  426. {
  427. _STARPU_MPI_LOG_IN();
  428. STARPU_ASSERT_MSG(public_req, "starpu_mpi_irecv needs a valid starpu_mpi_req");
  429. // // We check if a tag is defined for the data handle, if not,
  430. // // we define the one given for the communication.
  431. // // A tag is necessary for the internal mpi engine.
  432. // int tag = starpu_data_get_tag(data_handle);
  433. // if (tag == -1)
  434. // starpu_data_set_tag(data_handle, mpi_tag);
  435. struct _starpu_mpi_req *req;
  436. req = _starpu_mpi_irecv_common(data_handle, source, mpi_tag, comm, 0, NULL, NULL, 1, 0, 0);
  437. STARPU_ASSERT_MSG(req, "Invalid return for _starpu_mpi_irecv_common");
  438. *public_req = req;
  439. _STARPU_MPI_LOG_OUT();
  440. return 0;
  441. }
  442. int starpu_mpi_irecv_detached(starpu_data_handle_t data_handle, int source, int mpi_tag, MPI_Comm comm, void (*callback)(void *), void *arg)
  443. {
  444. _STARPU_MPI_LOG_IN();
  445. // // We check if a tag is defined for the data handle, if not,
  446. // // we define the one given for the communication.
  447. // // A tag is necessary for the internal mpi engine.
  448. // int tag = starpu_data_get_tag(data_handle);
  449. // if (tag == -1)
  450. // starpu_data_set_tag(data_handle, mpi_tag);
  451. _starpu_mpi_irecv_common(data_handle, source, mpi_tag, comm, 1, callback, arg, 1, 0, 0);
  452. _STARPU_MPI_LOG_OUT();
  453. return 0;
  454. }
  455. int starpu_mpi_irecv_detached_sequential_consistency(starpu_data_handle_t data_handle, int source, int mpi_tag, MPI_Comm comm, void (*callback)(void *), void *arg, int sequential_consistency)
  456. {
  457. _STARPU_MPI_LOG_IN();
  458. // // We check if a tag is defined for the data handle, if not,
  459. // // we define the one given for the communication.
  460. // // A tag is necessary for the internal mpi engine.
  461. // int tag = starpu_data_get_tag(data_handle);
  462. // if (tag == -1)
  463. // starpu_data_set_tag(data_handle, mpi_tag);
  464. _starpu_mpi_irecv_common(data_handle, source, mpi_tag, comm, 1, callback, arg, sequential_consistency, 0, 0);
  465. _STARPU_MPI_LOG_OUT();
  466. return 0;
  467. }
  468. int starpu_mpi_recv(starpu_data_handle_t data_handle, int source, int mpi_tag, MPI_Comm comm, MPI_Status *status)
  469. {
  470. starpu_mpi_req req;
  471. _STARPU_MPI_LOG_IN();
  472. // // We check if a tag is defined for the data handle, if not,
  473. // // we define the one given for the communication.
  474. // // A tag is necessary for the internal mpi engine.
  475. // int tag = starpu_data_get_tag(data_handle);
  476. // if (tag == -1)
  477. // starpu_data_set_tag(data_handle, mpi_tag);
  478. starpu_mpi_irecv(data_handle, &req, source, mpi_tag, comm);
  479. starpu_mpi_wait(&req, status);
  480. _STARPU_MPI_LOG_OUT();
  481. return 0;
  482. }
  483. /********************************************************/
  484. /* */
  485. /* Wait functionalities */
  486. /* */
  487. /********************************************************/
  488. static void _starpu_mpi_wait_func(struct _starpu_mpi_req *waiting_req)
  489. {
  490. _STARPU_MPI_LOG_IN();
  491. /* Which is the mpi request we are waiting for ? */
  492. struct _starpu_mpi_req *req = waiting_req->other_request;
  493. TRACE_MPI_UWAIT_BEGIN(req->srcdst, req->mpi_tag);
  494. req->ret = MPI_Wait(&req->request, waiting_req->status);
  495. STARPU_ASSERT_MSG(req->ret == MPI_SUCCESS, "MPI_Wait returning %d", req->ret);
  496. TRACE_MPI_UWAIT_END(req->srcdst, req->mpi_tag);
  497. _starpu_mpi_handle_request_termination(req);
  498. _STARPU_MPI_LOG_OUT();
  499. }
  500. int starpu_mpi_wait(starpu_mpi_req *public_req, MPI_Status *status)
  501. {
  502. _STARPU_MPI_LOG_IN();
  503. int ret;
  504. struct _starpu_mpi_req *waiting_req = malloc(sizeof(struct _starpu_mpi_req));
  505. _starpu_mpi_request_init(waiting_req);
  506. STARPU_ASSERT_MSG(waiting_req, "Allocation failed");
  507. struct _starpu_mpi_req *req = *public_req;
  508. _STARPU_MPI_INC_POSTED_REQUESTS(1);
  509. /* We cannot try to complete a MPI request that was not actually posted
  510. * to MPI yet. */
  511. STARPU_PTHREAD_MUTEX_LOCK(&(req->req_mutex));
  512. while (!(req->submitted))
  513. STARPU_PTHREAD_COND_WAIT(&(req->req_cond), &(req->req_mutex));
  514. STARPU_PTHREAD_MUTEX_UNLOCK(&(req->req_mutex));
  515. /* Initialize the request structure */
  516. _starpu_mpi_request_init(waiting_req);
  517. waiting_req->status = status;
  518. waiting_req->other_request = req;
  519. waiting_req->func = _starpu_mpi_wait_func;
  520. waiting_req->request_type = WAIT_REQ;
  521. _starpu_mpi_submit_new_mpi_request(waiting_req);
  522. /* We wait for the MPI request to finish */
  523. STARPU_PTHREAD_MUTEX_LOCK(&req->req_mutex);
  524. while (!req->completed)
  525. STARPU_PTHREAD_COND_WAIT(&req->req_cond, &req->req_mutex);
  526. STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
  527. ret = req->ret;
  528. /* The internal request structure was automatically allocated */
  529. *public_req = NULL;
  530. free(req);
  531. free(waiting_req);
  532. _STARPU_MPI_LOG_OUT();
  533. return ret;
  534. }
  535. /********************************************************/
  536. /* */
  537. /* Test functionalities */
  538. /* */
  539. /********************************************************/
  540. static void _starpu_mpi_test_func(struct _starpu_mpi_req *testing_req)
  541. {
  542. _STARPU_MPI_LOG_IN();
  543. /* Which is the mpi request we are testing for ? */
  544. struct _starpu_mpi_req *req = testing_req->other_request;
  545. _STARPU_MPI_DEBUG(2, "Test request %p type %s tag %d src %d data %p ptr %p datatype '%s' count %d user_datatype %d \n",
  546. req, _starpu_mpi_request_type(req->request_type), req->mpi_tag, req->srcdst, req->data_handle, req->ptr, _starpu_mpi_datatype(req->datatype), (int)req->count, req->user_datatype);
  547. TRACE_MPI_UTESTING_BEGIN(req->srcdst, req->mpi_tag);
  548. req->ret = MPI_Test(&req->request, testing_req->flag, testing_req->status);
  549. STARPU_ASSERT_MSG(req->ret == MPI_SUCCESS, "MPI_Test returning %d", req->ret);
  550. TRACE_MPI_UTESTING_END(req->srcdst, req->mpi_tag);
  551. if (*testing_req->flag)
  552. {
  553. testing_req->ret = req->ret;
  554. _starpu_mpi_handle_request_termination(req);
  555. }
  556. STARPU_PTHREAD_MUTEX_LOCK(&testing_req->req_mutex);
  557. testing_req->completed = 1;
  558. STARPU_PTHREAD_COND_SIGNAL(&testing_req->req_cond);
  559. STARPU_PTHREAD_MUTEX_UNLOCK(&testing_req->req_mutex);
  560. _STARPU_MPI_LOG_OUT();
  561. }
  562. int starpu_mpi_test(starpu_mpi_req *public_req, int *flag, MPI_Status *status)
  563. {
  564. _STARPU_MPI_LOG_IN();
  565. int ret = 0;
  566. STARPU_ASSERT_MSG(public_req, "starpu_mpi_test needs a valid starpu_mpi_req");
  567. struct _starpu_mpi_req *req = *public_req;
  568. STARPU_ASSERT_MSG(!req->detached, "MPI_Test cannot be called on a detached request");
  569. STARPU_PTHREAD_MUTEX_LOCK(&req->req_mutex);
  570. unsigned submitted = req->submitted;
  571. STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
  572. if (submitted)
  573. {
  574. struct _starpu_mpi_req *testing_req = malloc(sizeof(struct _starpu_mpi_req));
  575. STARPU_ASSERT_MSG(testing_req, "allocation failed");
  576. _starpu_mpi_request_init(testing_req);
  577. /* Initialize the request structure */
  578. STARPU_PTHREAD_MUTEX_INIT(&(testing_req->req_mutex), NULL);
  579. STARPU_PTHREAD_COND_INIT(&(testing_req->req_cond), NULL);
  580. testing_req->flag = flag;
  581. testing_req->status = status;
  582. testing_req->other_request = req;
  583. testing_req->func = _starpu_mpi_test_func;
  584. testing_req->completed = 0;
  585. testing_req->request_type = TEST_REQ;
  586. _STARPU_MPI_INC_POSTED_REQUESTS(1);
  587. _starpu_mpi_submit_new_mpi_request(testing_req);
  588. /* We wait for the test request to finish */
  589. STARPU_PTHREAD_MUTEX_LOCK(&(testing_req->req_mutex));
  590. while (!(testing_req->completed))
  591. STARPU_PTHREAD_COND_WAIT(&(testing_req->req_cond), &(testing_req->req_mutex));
  592. STARPU_PTHREAD_MUTEX_UNLOCK(&(testing_req->req_mutex));
  593. ret = testing_req->ret;
  594. if (*(testing_req->flag))
  595. {
  596. /* The request was completed so we free the internal
  597. * request structure which was automatically allocated
  598. * */
  599. *public_req = NULL;
  600. free(req);
  601. }
  602. free(testing_req);
  603. }
  604. else
  605. {
  606. *flag = 0;
  607. }
  608. _STARPU_MPI_LOG_OUT();
  609. return ret;
  610. }
  611. /********************************************************/
  612. /* */
  613. /* Barrier functionalities */
  614. /* */
  615. /********************************************************/
  616. static void _starpu_mpi_barrier_func(struct _starpu_mpi_req *barrier_req)
  617. {
  618. _STARPU_MPI_LOG_IN();
  619. barrier_req->ret = MPI_Barrier(barrier_req->comm);
  620. STARPU_ASSERT_MSG(barrier_req->ret == MPI_SUCCESS, "MPI_Barrier returning %d", barrier_req->ret);
  621. _starpu_mpi_handle_request_termination(barrier_req);
  622. _STARPU_MPI_LOG_OUT();
  623. }
  624. int starpu_mpi_barrier(MPI_Comm comm)
  625. {
  626. _STARPU_MPI_LOG_IN();
  627. int ret;
  628. struct _starpu_mpi_req *barrier_req = malloc(sizeof(struct _starpu_mpi_req));
  629. STARPU_ASSERT_MSG(barrier_req, "allocation failed");
  630. _starpu_mpi_request_init(barrier_req);
  631. /* First wait for *both* all tasks and MPI requests to finish, in case
  632. * some tasks generate MPI requests, MPI requests generate tasks, etc.
  633. */
  634. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  635. STARPU_ASSERT_MSG(!barrier_running, "Concurrent starpu_mpi_barrier is not implemented, even on different communicators");
  636. barrier_running = 1;
  637. do
  638. {
  639. while (posted_requests)
  640. /* Wait for all current MPI requests to finish */
  641. STARPU_PTHREAD_COND_WAIT(&cond_finished, &mutex);
  642. /* No current request, clear flag */
  643. newer_requests = 0;
  644. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  645. /* Now wait for all tasks */
  646. starpu_task_wait_for_all();
  647. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  648. /* Check newer_requests again, in case some MPI requests
  649. * triggered by tasks completed and triggered tasks between
  650. * wait_for_all finished and we take the lock */
  651. } while (posted_requests || newer_requests);
  652. barrier_running = 0;
  653. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  654. /* Initialize the request structure */
  655. STARPU_PTHREAD_MUTEX_INIT(&(barrier_req->req_mutex), NULL);
  656. STARPU_PTHREAD_COND_INIT(&(barrier_req->req_cond), NULL);
  657. barrier_req->func = _starpu_mpi_barrier_func;
  658. barrier_req->request_type = BARRIER_REQ;
  659. barrier_req->comm = comm;
  660. _STARPU_MPI_INC_POSTED_REQUESTS(1);
  661. _starpu_mpi_submit_new_mpi_request(barrier_req);
  662. /* We wait for the MPI request to finish */
  663. STARPU_PTHREAD_MUTEX_LOCK(&barrier_req->req_mutex);
  664. while (!barrier_req->completed)
  665. STARPU_PTHREAD_COND_WAIT(&barrier_req->req_cond, &barrier_req->req_mutex);
  666. STARPU_PTHREAD_MUTEX_UNLOCK(&barrier_req->req_mutex);
  667. ret = barrier_req->ret;
  668. free(barrier_req);
  669. _STARPU_MPI_LOG_OUT();
  670. return ret;
  671. }
  672. /********************************************************/
  673. /* */
  674. /* Progression */
  675. /* */
  676. /********************************************************/
  677. #ifdef STARPU_VERBOSE
  678. static char *_starpu_mpi_request_type(enum _starpu_mpi_request_type request_type)
  679. {
  680. switch (request_type)
  681. {
  682. case SEND_REQ: return "SEND_REQ";
  683. case RECV_REQ: return "RECV_REQ";
  684. case WAIT_REQ: return "WAIT_REQ";
  685. case TEST_REQ: return "TEST_REQ";
  686. case BARRIER_REQ: return "BARRIER_REQ";
  687. case UNKNOWN_REQ: return "UNSET_REQ";
  688. default: return "unknown request type";
  689. }
  690. }
  691. #endif
  692. static void _starpu_mpi_handle_request_termination(struct _starpu_mpi_req *req)
  693. {
  694. int ret;
  695. _STARPU_MPI_LOG_IN();
  696. _STARPU_MPI_DEBUG(2, "complete MPI request %p type %s tag %d src %d data %p ptr %p datatype '%s' count %d user_datatype %d internal_req %p\n",
  697. req, _starpu_mpi_request_type(req->request_type), req->mpi_tag, req->srcdst, req->data_handle, req->ptr,
  698. _starpu_mpi_datatype(req->datatype), (int)req->count, req->user_datatype, req->internal_req);
  699. if (req->internal_req)
  700. {
  701. struct _starpu_mpi_copy_handle *chandle = find_chandle(req->mpi_tag, req->srcdst);
  702. STARPU_ASSERT_MSG(chandle, "Could not find a copy data handle with the tag %d and the node %d\n", req->mpi_tag, req->srcdst);
  703. _STARPU_MPI_DEBUG(3, "Handling deleting of copy_handle structure from the hashmap..\n");
  704. delete_chandle(chandle);
  705. free(chandle);
  706. }
  707. else
  708. {
  709. if (req->request_type == RECV_REQ || req->request_type == SEND_REQ)
  710. {
  711. if (req->user_datatype == 1)
  712. {
  713. if (req->request_type == SEND_REQ)
  714. {
  715. // We need to make sure the communication for sending the size
  716. // has completed, as MPI can re-order messages, let's call
  717. // MPI_Wait to make sure data have been sent
  718. ret = MPI_Wait(&req->size_req, MPI_STATUS_IGNORE);
  719. STARPU_ASSERT_MSG(ret == MPI_SUCCESS, "MPI_Wait returning %d", ret);
  720. free(req->ptr);
  721. }
  722. if (req->request_type == RECV_REQ)
  723. {
  724. // req->ptr is freed by starpu_data_unpack
  725. starpu_data_unpack(req->data_handle, req->ptr, req->count);
  726. }
  727. }
  728. else
  729. {
  730. _starpu_mpi_handle_free_datatype(req->data_handle, &req->datatype);
  731. }
  732. }
  733. }
  734. if (req->data_handle)
  735. starpu_data_release(req->data_handle);
  736. if (req->envelope)
  737. {
  738. free(req->envelope);
  739. req->envelope = NULL;
  740. }
  741. /* Execute the specified callback, if any */
  742. if (req->callback)
  743. req->callback(req->callback_arg);
  744. /* tell anyone potentially waiting on the request that it is
  745. * terminated now */
  746. STARPU_PTHREAD_MUTEX_LOCK(&req->req_mutex);
  747. req->completed = 1;
  748. STARPU_PTHREAD_COND_BROADCAST(&req->req_cond);
  749. STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
  750. _STARPU_MPI_LOG_OUT();
  751. }
  752. struct _starpu_mpi_copy_cb_args
  753. {
  754. starpu_data_handle_t data_handle;
  755. starpu_data_handle_t copy_handle;
  756. struct _starpu_mpi_req *req;
  757. void *buffer;
  758. };
  759. static void _starpu_mpi_copy_cb(void* arg)
  760. {
  761. struct _starpu_mpi_copy_cb_args *args = arg;
  762. // We store in the application request the internal MPI
  763. // request so that it can be used by starpu_mpi_wait
  764. args->req->request = args->req->internal_req->request;
  765. args->req->submitted = 1;
  766. if (args->buffer)
  767. {
  768. /* Data has been received as a raw memory, it has to be unpacked */
  769. struct starpu_data_interface_ops *itf_src = starpu_data_get_interface_ops(args->copy_handle);
  770. struct starpu_data_interface_ops *itf_dst = starpu_data_get_interface_ops(args->data_handle);
  771. itf_dst->unpack_data(args->data_handle, 0, args->buffer, itf_src->get_size(args->copy_handle));
  772. free(args->buffer);
  773. }
  774. else
  775. {
  776. struct starpu_data_interface_ops *itf = starpu_data_get_interface_ops(args->copy_handle);
  777. void* itf_src = starpu_data_get_interface_on_node(args->copy_handle, STARPU_MAIN_RAM);
  778. void* itf_dst = starpu_data_get_interface_on_node(args->data_handle, STARPU_MAIN_RAM);
  779. if (!itf->copy_methods->ram_to_ram)
  780. {
  781. _STARPU_MPI_DEBUG(3, "Initiating any_to_any copy..\n");
  782. itf->copy_methods->any_to_any(itf_src, 0, itf_dst, 0, NULL);
  783. }
  784. else
  785. {
  786. _STARPU_MPI_DEBUG(3, "Initiating ram_to_ram copy..\n");
  787. itf->copy_methods->ram_to_ram(itf_src, 0, itf_dst, 0);
  788. }
  789. }
  790. _STARPU_MPI_DEBUG(3, "Done, handling release of copy_handle..\n");
  791. starpu_data_release(args->copy_handle);
  792. _STARPU_MPI_DEBUG(3, "Done, handling unregister of copy_handle..\n");
  793. starpu_data_unregister_submit(args->copy_handle);
  794. _STARPU_MPI_DEBUG(3, "Done, handling request %p termination of the already received request\n",args->req);
  795. // If the request is detached, we need to call _starpu_mpi_handle_request_termination
  796. // as it will not be called automatically as the request is not in the list detached_requests
  797. if (args->req->detached)
  798. _starpu_mpi_handle_request_termination(args->req);
  799. // else: If the request is not detached its termination will
  800. // be handled when calling starpu_mpi_wait
  801. free(args);
  802. }
  803. static void _starpu_mpi_submit_new_mpi_request(void *arg)
  804. {
  805. _STARPU_MPI_LOG_IN();
  806. struct _starpu_mpi_req *req = arg;
  807. _STARPU_MPI_INC_POSTED_REQUESTS(-1);
  808. _STARPU_MPI_DEBUG(3, "calling _starpu_mpi_submit_new_mpi_request with req %p srcdst %d tag %d and type %s\n", req, req->srcdst, req->mpi_tag, _starpu_mpi_request_type(req->request_type));
  809. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  810. if (req->request_type == RECV_REQ)
  811. {
  812. /* Case : the request is the internal receive request submitted by StarPU-MPI to receive
  813. * incoming data without a matching pending receive already submitted by the application.
  814. * We immediately allocate the pointer associated to the data_handle, and pushing it into
  815. * the list of new_requests, so as the real MPI request can be submitted before the next
  816. * submission of the envelope-catching request. */
  817. if (req->is_internal_req)
  818. {
  819. _starpu_mpi_handle_allocate_datatype(req->data_handle, &req->datatype, &req->user_datatype);
  820. if (req->user_datatype == 0)
  821. {
  822. req->count = 1;
  823. req->ptr = starpu_data_get_local_ptr(req->data_handle);
  824. }
  825. else
  826. {
  827. STARPU_ASSERT(req->count);
  828. req->ptr = malloc(req->count);
  829. STARPU_ASSERT_MSG(req->ptr, "cannot allocate message of size %ld\n", req->count);
  830. }
  831. _STARPU_MPI_DEBUG(3, "Pushing internal starpu_mpi_irecv request %p type %s tag %d src %d data %p ptr %p datatype '%s' count %d user_datatype %d \n", req, _starpu_mpi_request_type(req->request_type), req->mpi_tag, req->srcdst, req->data_handle, req->ptr, _starpu_mpi_datatype(req->datatype), (int)req->count, req->user_datatype);
  832. _starpu_mpi_req_list_push_front(new_requests, req);
  833. /* inform the starpu mpi thread that the request has beenbe pushed in the new_requests list */
  834. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  835. STARPU_PTHREAD_MUTEX_LOCK(&req->posted_mutex);
  836. req->posted = 1;
  837. STARPU_PTHREAD_COND_BROADCAST(&req->posted_cond);
  838. STARPU_PTHREAD_MUTEX_UNLOCK(&req->posted_mutex);
  839. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  840. }
  841. else
  842. {
  843. /* test whether the receive request has already been submitted internally by StarPU-MPI*/
  844. struct _starpu_mpi_copy_handle *chandle = find_chandle(req->mpi_tag, req->srcdst);
  845. /* Case : the request has already been submitted internally by StarPU.
  846. * We'll asynchronously ask a Read permission over the temporary handle, so as when
  847. * the internal receive will be over, the _starpu_mpi_copy_cb function will be called to
  848. * bring the data back to the original data handle associated to the request.*/
  849. if (chandle)
  850. {
  851. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  852. STARPU_PTHREAD_MUTEX_LOCK(&(chandle->req_mutex));
  853. while (!(chandle->req_ready))
  854. STARPU_PTHREAD_COND_WAIT(&(chandle->req_cond), &(chandle->req_mutex));
  855. STARPU_PTHREAD_MUTEX_UNLOCK(&(chandle->req_mutex));
  856. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  857. _STARPU_MPI_DEBUG(3, "The RECV request %p with tag %d has already been received, copying previously received data into handle's pointer..\n", req, req->mpi_tag);
  858. STARPU_ASSERT(req->data_handle != chandle->handle);
  859. req->internal_req = chandle->req;
  860. struct _starpu_mpi_copy_cb_args *cb_args = malloc(sizeof(struct _starpu_mpi_copy_cb_args));
  861. cb_args->data_handle = req->data_handle;
  862. cb_args->copy_handle = chandle->handle;
  863. cb_args->buffer = chandle->buffer;
  864. cb_args->req = req;
  865. _STARPU_MPI_DEBUG(3, "Calling data_acquire_cb on starpu_mpi_copy_cb..\n");
  866. starpu_data_acquire_cb(chandle->handle,STARPU_R,_starpu_mpi_copy_cb,(void*) cb_args);
  867. }
  868. /* Case : a classic receive request with no send received earlier than expected.
  869. * We just add the pending receive request to the requests' hashmap. */
  870. else
  871. {
  872. _STARPU_MPI_DEBUG(3, "Adding the pending receive request %p (srcdst %d tag %d) into the request hashmap\n", req, req->srcdst, req->mpi_tag);
  873. add_app_req(req);
  874. }
  875. }
  876. }
  877. else
  878. {
  879. _starpu_mpi_req_list_push_front(new_requests, req);
  880. _STARPU_MPI_DEBUG(3, "Pushing new request %p type %s tag %d src %d data %p ptr %p datatype '%s' count %d user_datatype %d \n",
  881. req, _starpu_mpi_request_type(req->request_type), req->mpi_tag, req->srcdst, req->data_handle, req->ptr, _starpu_mpi_datatype(req->datatype), (int)req->count, req->user_datatype);
  882. }
  883. newer_requests = 1;
  884. STARPU_PTHREAD_COND_BROADCAST(&cond_progression);
  885. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  886. _STARPU_MPI_LOG_OUT();
  887. }
  888. #ifdef STARPU_MPI_ACTIVITY
  889. static unsigned _starpu_mpi_progression_hook_func(void *arg STARPU_ATTRIBUTE_UNUSED)
  890. {
  891. unsigned may_block = 1;
  892. STARPU_PTHREAD_MUTEX_LOCK(&detached_requests_mutex);
  893. if (!_starpu_mpi_req_list_empty(detached_requests))
  894. {
  895. STARPU_PTHREAD_MUTEX_UNLOCK(&detached_requests_mutex);
  896. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  897. STARPU_PTHREAD_COND_SIGNAL(&cond_progression);
  898. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  899. may_block = 0;
  900. }
  901. else
  902. STARPU_PTHREAD_MUTEX_UNLOCK(&detached_requests_mutex);
  903. return may_block;
  904. }
  905. #endif /* STARPU_MPI_ACTIVITY */
  906. static void _starpu_mpi_test_detached_requests(void)
  907. {
  908. _STARPU_MPI_LOG_IN();
  909. int flag;
  910. MPI_Status status;
  911. struct _starpu_mpi_req *req, *next_req;
  912. STARPU_PTHREAD_MUTEX_LOCK(&detached_requests_mutex);
  913. for (req = _starpu_mpi_req_list_begin(detached_requests);
  914. req != _starpu_mpi_req_list_end(detached_requests);
  915. req = next_req)
  916. {
  917. next_req = _starpu_mpi_req_list_next(req);
  918. STARPU_PTHREAD_MUTEX_UNLOCK(&detached_requests_mutex);
  919. //_STARPU_MPI_DEBUG(3, "Test detached request %p - mpitag %d - TYPE %s %d\n", &req->request, req->mpi_tag, _starpu_mpi_request_type(req->request_type), req->srcdst);
  920. req->ret = MPI_Test(&req->request, &flag, &status);
  921. STARPU_ASSERT_MSG(req->ret == MPI_SUCCESS, "MPI_Test returning %d", req->ret);
  922. if (flag)
  923. {
  924. if (req->request_type == RECV_REQ)
  925. {
  926. TRACE_MPI_IRECV_COMPLETE_BEGIN(req->srcdst, req->mpi_tag);
  927. }
  928. else if (req->request_type == SEND_REQ)
  929. {
  930. TRACE_MPI_ISEND_COMPLETE_BEGIN(req->srcdst, req->mpi_tag, 0);
  931. }
  932. _starpu_mpi_handle_request_termination(req);
  933. if (req->request_type == RECV_REQ)
  934. {
  935. TRACE_MPI_IRECV_COMPLETE_END(req->srcdst, req->mpi_tag);
  936. }
  937. else if (req->request_type == SEND_REQ)
  938. {
  939. TRACE_MPI_ISEND_COMPLETE_END(req->srcdst, req->mpi_tag, 0);
  940. }
  941. }
  942. STARPU_PTHREAD_MUTEX_LOCK(&detached_requests_mutex);
  943. if (flag)
  944. {
  945. _starpu_mpi_req_list_erase(detached_requests, req);
  946. #ifdef STARPU_DEVEL
  947. #warning FIXME: when do we free internal requests
  948. #endif
  949. if (!req->is_internal_req)
  950. free(req);
  951. }
  952. }
  953. STARPU_PTHREAD_MUTEX_UNLOCK(&detached_requests_mutex);
  954. _STARPU_MPI_LOG_OUT();
  955. }
  956. static void _starpu_mpi_handle_detached_request(struct _starpu_mpi_req *req)
  957. {
  958. if (req->detached)
  959. {
  960. /* put the submitted request into the list of pending requests
  961. * so that it can be handled by the progression mechanisms */
  962. STARPU_PTHREAD_MUTEX_LOCK(&detached_requests_mutex);
  963. _starpu_mpi_req_list_push_front(detached_requests, req);
  964. STARPU_PTHREAD_MUTEX_UNLOCK(&detached_requests_mutex);
  965. starpu_wake_all_blocked_workers();
  966. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  967. STARPU_PTHREAD_COND_SIGNAL(&cond_progression);
  968. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  969. }
  970. }
  971. static void _starpu_mpi_handle_new_request(struct _starpu_mpi_req *req)
  972. {
  973. _STARPU_MPI_LOG_IN();
  974. STARPU_ASSERT_MSG(req, "Invalid request");
  975. /* submit the request to MPI */
  976. _STARPU_MPI_DEBUG(2, "Handling new request %p type %s tag %d src %d data %p ptr %p datatype '%s' count %d user_datatype %d \n",
  977. req, _starpu_mpi_request_type(req->request_type), req->mpi_tag, req->srcdst, req->data_handle, req->ptr, _starpu_mpi_datatype(req->datatype), (int)req->count, req->user_datatype);
  978. req->func(req);
  979. _STARPU_MPI_LOG_OUT();
  980. }
  981. struct _starpu_mpi_argc_argv
  982. {
  983. int initialize_mpi;
  984. int *argc;
  985. char ***argv;
  986. };
  987. static void _starpu_mpi_print_thread_level_support(int thread_level, char *msg)
  988. {
  989. switch (thread_level)
  990. {
  991. case MPI_THREAD_SERIALIZED:
  992. {
  993. _STARPU_DISP("MPI%s MPI_THREAD_SERIALIZED; Multiple threads may make MPI calls, but only one at a time.\n", msg);
  994. break;
  995. }
  996. case MPI_THREAD_FUNNELED:
  997. {
  998. _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);
  999. break;
  1000. }
  1001. case MPI_THREAD_SINGLE:
  1002. {
  1003. _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);
  1004. break;
  1005. }
  1006. }
  1007. }
  1008. static void *_starpu_mpi_progress_thread_func(void *arg)
  1009. {
  1010. struct _starpu_mpi_argc_argv *argc_argv = (struct _starpu_mpi_argc_argv *) arg;
  1011. int rank, worldsize;
  1012. if (argc_argv->initialize_mpi)
  1013. {
  1014. int thread_support;
  1015. _STARPU_DEBUG("Calling MPI_Init_thread\n");
  1016. if (MPI_Init_thread(argc_argv->argc, argc_argv->argv, MPI_THREAD_SERIALIZED, &thread_support) != MPI_SUCCESS)
  1017. {
  1018. _STARPU_ERROR("MPI_Init_thread failed\n");
  1019. }
  1020. _starpu_mpi_print_thread_level_support(thread_support, "_Init_thread level =");
  1021. }
  1022. else
  1023. {
  1024. int provided;
  1025. MPI_Query_thread(&provided);
  1026. _starpu_mpi_print_thread_level_support(provided, " has been initialized with");
  1027. }
  1028. MPI_Comm_rank(MPI_COMM_WORLD, &rank);
  1029. MPI_Comm_size(MPI_COMM_WORLD, &worldsize);
  1030. MPI_Comm_set_errhandler(MPI_COMM_WORLD, MPI_ERRORS_RETURN);
  1031. {
  1032. TRACE_MPI_START(rank, worldsize);
  1033. #ifdef STARPU_USE_FXT
  1034. starpu_profiling_set_id(rank);
  1035. #endif //STARPU_USE_FXT
  1036. }
  1037. _starpu_mpi_add_sync_point_in_fxt();
  1038. _starpu_mpi_comm_amounts_init(MPI_COMM_WORLD);
  1039. _starpu_mpi_cache_init(MPI_COMM_WORLD);
  1040. {
  1041. int nb_nodes, k;
  1042. MPI_Comm_size(MPI_COMM_WORLD, &nb_nodes);
  1043. _starpu_mpi_app_req_hashmap = malloc(nb_nodes * sizeof(struct _starpu_mpi_req *));
  1044. for(k=0 ; k<nb_nodes ; k++) _starpu_mpi_app_req_hashmap[k] = NULL;
  1045. _starpu_mpi_copy_handle_hashmap = malloc(nb_nodes * sizeof(struct _starpu_mpi_copy_handle_hash_list *));
  1046. for(k=0 ; k<nb_nodes ; k++) _starpu_mpi_copy_handle_hashmap[k] = NULL;
  1047. }
  1048. /* notify the main thread that the progression thread is ready */
  1049. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  1050. running = 1;
  1051. STARPU_PTHREAD_COND_SIGNAL(&cond_progression);
  1052. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  1053. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  1054. struct _starpu_mpi_envelope *recv_env = calloc(1,sizeof(struct _starpu_mpi_envelope));
  1055. int header_req_submitted = 0;
  1056. while (running || posted_requests || !(_starpu_mpi_req_list_empty(new_requests)) || !(_starpu_mpi_req_list_empty(detached_requests)))
  1057. {
  1058. /* shall we block ? */
  1059. unsigned block = _starpu_mpi_req_list_empty(new_requests) && (_starpu_mpi_app_req_hashmap_count == 0);
  1060. #ifndef STARPU_MPI_ACTIVITY
  1061. STARPU_PTHREAD_MUTEX_LOCK(&detached_requests_mutex);
  1062. block = block && _starpu_mpi_req_list_empty(detached_requests);
  1063. STARPU_PTHREAD_MUTEX_UNLOCK(&detached_requests_mutex);
  1064. #endif /* STARPU_MPI_ACTIVITY */
  1065. if (block)
  1066. {
  1067. _STARPU_MPI_DEBUG(3, "NO MORE REQUESTS TO HANDLE\n");
  1068. TRACE_MPI_SLEEP_BEGIN();
  1069. if (barrier_running)
  1070. /* Tell mpi_barrier */
  1071. STARPU_PTHREAD_COND_SIGNAL(&cond_finished);
  1072. STARPU_PTHREAD_COND_WAIT(&cond_progression, &mutex);
  1073. TRACE_MPI_SLEEP_END();
  1074. }
  1075. /* get one request */
  1076. struct _starpu_mpi_req *req;
  1077. while (!_starpu_mpi_req_list_empty(new_requests))
  1078. {
  1079. req = _starpu_mpi_req_list_pop_back(new_requests);
  1080. /* handling a request is likely to block for a while
  1081. * (on a sync_data_with_mem call), we want to let the
  1082. * application submit requests in the meantime, so we
  1083. * release the lock. */
  1084. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  1085. _starpu_mpi_handle_new_request(req);
  1086. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  1087. }
  1088. /* If there is no currently submitted header_req submitted to catch envelopes from senders, and there is some pending receive
  1089. * requests in our side, we resubmit a header request. */
  1090. MPI_Request header_req;
  1091. if ((_starpu_mpi_app_req_hashmap_count > 0) && (header_req_submitted == 0))// && (HASH_COUNT(_starpu_mpi_copy_handle_hashmap) == 0))
  1092. {
  1093. _STARPU_MPI_DEBUG(3, "Posting a receive to get a data envelop\n");
  1094. MPI_Irecv(recv_env, sizeof(struct _starpu_mpi_envelope), MPI_BYTE, MPI_ANY_SOURCE, _starpu_mpi_tag, MPI_COMM_WORLD, &header_req);
  1095. header_req_submitted = 1;
  1096. }
  1097. /* test whether there are some terminated "detached request" */
  1098. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  1099. _starpu_mpi_test_detached_requests();
  1100. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  1101. if (header_req_submitted == 1)
  1102. {
  1103. int flag,res;
  1104. MPI_Status status;
  1105. _STARPU_MPI_DEBUG(4, "Test of header_req\n");
  1106. /* test whether an envelope has arrived. */
  1107. res = MPI_Test(&header_req, &flag, &status);
  1108. STARPU_ASSERT(res == MPI_SUCCESS);
  1109. if (flag)
  1110. {
  1111. _STARPU_MPI_DEBUG(3, "Searching for application request with tag %d and source %d (size %ld)\n", recv_env->mpi_tag, status.MPI_SOURCE, recv_env->size);
  1112. struct _starpu_mpi_req *found_req = find_app_req(recv_env->mpi_tag, status.MPI_SOURCE);
  1113. /* Case : a data will arrive before the matching receive has been submitted in our side of the application.
  1114. * We will allow a temporary handle to store the incoming data, by submitting a starpu_mpi_irecv_detached
  1115. * on this handle, and register this so as the StarPU-MPI layer can remember it.*/
  1116. if (!found_req)
  1117. {
  1118. _STARPU_MPI_DEBUG(3, "Request with tag %d and source %d not found, creating a copy_handle to receive incoming data..\n", recv_env->mpi_tag, status.MPI_SOURCE);
  1119. starpu_data_handle_t data_handle = NULL;
  1120. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  1121. data_handle = starpu_data_get_data_handle_from_tag(recv_env->mpi_tag);
  1122. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  1123. struct _starpu_mpi_copy_handle* chandle = calloc(1, sizeof(struct _starpu_mpi_copy_handle));
  1124. STARPU_ASSERT(chandle);
  1125. STARPU_PTHREAD_MUTEX_INIT(&chandle->req_mutex, NULL);
  1126. STARPU_PTHREAD_COND_INIT(&chandle->req_cond, NULL);
  1127. chandle->mpi_tag = recv_env->mpi_tag;
  1128. chandle->env = recv_env;
  1129. chandle->source = status.MPI_SOURCE;
  1130. if (data_handle)
  1131. {
  1132. chandle->buffer = NULL;
  1133. starpu_data_register_same(&chandle->handle, data_handle);
  1134. add_chandle(chandle);
  1135. }
  1136. else
  1137. {
  1138. /* The application has not registered yet a data with the tag,
  1139. * we are going to receive the data as a raw memory, and give it
  1140. * to the application when it post a receive for this tag
  1141. */
  1142. _STARPU_MPI_DEBUG(20, "Posting a receive for a data of size %d which has not yet been registered\n", (int)chandle->env->size);
  1143. chandle->buffer = malloc(chandle->env->size);
  1144. starpu_vector_data_register(&chandle->handle, STARPU_MAIN_RAM, (uintptr_t) chandle->buffer, chandle->env->size, 1);
  1145. add_chandle(chandle);
  1146. }
  1147. _STARPU_MPI_DEBUG(20, "Posting internal detached irecv on copy_handle with tag %d from src %d ..\n", chandle->mpi_tag, status.MPI_SOURCE);
  1148. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  1149. chandle->req = _starpu_mpi_irecv_common(chandle->handle, status.MPI_SOURCE, chandle->mpi_tag, MPI_COMM_WORLD, 1, NULL, NULL, 1, 1, recv_env->size);
  1150. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  1151. // We wait until the request is pushed in the
  1152. // new_request list, that ensures that the next loop
  1153. // will call _starpu_mpi_handle_new_request
  1154. // on the request and post the corresponding mpi_irecv,
  1155. // otherwise, it may lead to read data as envelop
  1156. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  1157. STARPU_PTHREAD_MUTEX_LOCK(&(chandle->req->posted_mutex));
  1158. while (!(chandle->req->posted))
  1159. STARPU_PTHREAD_COND_WAIT(&(chandle->req->posted_cond), &(chandle->req->posted_mutex));
  1160. STARPU_PTHREAD_MUTEX_UNLOCK(&(chandle->req->posted_mutex));
  1161. STARPU_PTHREAD_MUTEX_LOCK(&chandle->req_mutex);
  1162. chandle->req_ready = 1;
  1163. STARPU_PTHREAD_COND_BROADCAST(&chandle->req_cond);
  1164. STARPU_PTHREAD_MUTEX_UNLOCK(&chandle->req_mutex);
  1165. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  1166. }
  1167. /* Case : a matching receive has been found for the incoming data, we handle the correct allocation of the pointer associated to
  1168. * the data handle, then submit the corresponding receive with _starpu_mpi_handle_new_request. */
  1169. else
  1170. {
  1171. _STARPU_MPI_DEBUG(3, "A matching receive has been found for the incoming data with tag %d\n", recv_env->mpi_tag);
  1172. delete_app_req(found_req);
  1173. _starpu_mpi_handle_allocate_datatype(found_req->data_handle, &found_req->datatype, &found_req->user_datatype);
  1174. if (found_req->user_datatype == 0)
  1175. {
  1176. found_req->count = 1;
  1177. found_req->ptr = starpu_data_get_local_ptr(found_req->data_handle);
  1178. }
  1179. else
  1180. {
  1181. found_req->count = recv_env->size;
  1182. found_req->ptr = malloc(found_req->count);
  1183. STARPU_ASSERT_MSG(found_req->ptr, "cannot allocate message of size %ld\n", found_req->count);
  1184. }
  1185. _STARPU_MPI_DEBUG(3, "Handling new request... \n");
  1186. /* handling a request is likely to block for a while
  1187. * (on a sync_data_with_mem call), we want to let the
  1188. * application submit requests in the meantime, so we
  1189. * release the lock. */
  1190. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  1191. _starpu_mpi_handle_new_request(found_req);
  1192. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  1193. }
  1194. header_req_submitted = 0;
  1195. }
  1196. else
  1197. {
  1198. _STARPU_MPI_DEBUG(4, "Nothing received, continue ..\n");
  1199. }
  1200. }
  1201. }
  1202. STARPU_ASSERT_MSG(_starpu_mpi_req_list_empty(detached_requests), "List of detached requests not empty");
  1203. STARPU_ASSERT_MSG(_starpu_mpi_req_list_empty(new_requests), "List of new requests not empty");
  1204. STARPU_ASSERT_MSG(posted_requests == 0, "Number of posted request is not zero");
  1205. STARPU_ASSERT_MSG(_starpu_mpi_app_req_hashmap_count == 0, "Number of receive requests left is not zero");
  1206. STARPU_ASSERT_MSG(_starpu_mpi_copy_handle_hashmap_count == 0, "Number of copy requests left is not zero");
  1207. if (argc_argv->initialize_mpi)
  1208. {
  1209. _STARPU_MPI_DEBUG(3, "Calling MPI_Finalize()\n");
  1210. MPI_Finalize();
  1211. }
  1212. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  1213. {
  1214. int n;
  1215. struct _starpu_mpi_copy_handle_hashlist *hashlist;
  1216. for(n=0 ; n<worldsize; n++)
  1217. {
  1218. for(hashlist=_starpu_mpi_copy_handle_hashmap[n]; hashlist != NULL; hashlist=hashlist->hh.next)
  1219. {
  1220. _starpu_mpi_copy_handle_list_delete(hashlist->list);
  1221. }
  1222. struct _starpu_mpi_copy_handle_hashlist *current, *tmp;
  1223. HASH_ITER(hh, _starpu_mpi_copy_handle_hashmap[n], current, tmp)
  1224. {
  1225. free(current);
  1226. HASH_DEL(_starpu_mpi_copy_handle_hashmap[n], current);
  1227. }
  1228. }
  1229. }
  1230. free(_starpu_mpi_app_req_hashmap);
  1231. free(_starpu_mpi_copy_handle_hashmap);
  1232. free(argc_argv);
  1233. free(recv_env);
  1234. return NULL;
  1235. }
  1236. /********************************************************/
  1237. /* */
  1238. /* (De)Initialization methods */
  1239. /* */
  1240. /********************************************************/
  1241. #ifdef STARPU_MPI_ACTIVITY
  1242. static int hookid = - 1;
  1243. #endif /* STARPU_MPI_ACTIVITY */
  1244. static void _starpu_mpi_add_sync_point_in_fxt(void)
  1245. {
  1246. #ifdef STARPU_USE_FXT
  1247. int rank;
  1248. int worldsize;
  1249. int ret;
  1250. MPI_Comm_rank(MPI_COMM_WORLD, &rank);
  1251. MPI_Comm_size(MPI_COMM_WORLD, &worldsize);
  1252. ret = MPI_Barrier(MPI_COMM_WORLD);
  1253. STARPU_ASSERT_MSG(ret == MPI_SUCCESS, "MPI_Barrier returning %d", ret);
  1254. /* We generate a "unique" key so that we can make sure that different
  1255. * FxT traces come from the same MPI run. */
  1256. int random_number;
  1257. /* XXX perhaps we don't want to generate a new seed if the application
  1258. * specified some reproductible behaviour ? */
  1259. if (rank == 0)
  1260. {
  1261. srand(time(NULL));
  1262. random_number = rand();
  1263. }
  1264. ret = MPI_Bcast(&random_number, 1, MPI_INT, 0, MPI_COMM_WORLD);
  1265. STARPU_ASSERT_MSG(ret == MPI_SUCCESS, "MPI_Bcast returning %d", ret);
  1266. TRACE_MPI_BARRIER(rank, worldsize, random_number);
  1267. _STARPU_MPI_DEBUG(3, "unique key %x\n", random_number);
  1268. #endif
  1269. }
  1270. static
  1271. int _starpu_mpi_initialize(int *argc, char ***argv, int initialize_mpi)
  1272. {
  1273. STARPU_PTHREAD_MUTEX_INIT(&mutex, NULL);
  1274. STARPU_PTHREAD_COND_INIT(&cond_progression, NULL);
  1275. STARPU_PTHREAD_COND_INIT(&cond_finished, NULL);
  1276. new_requests = _starpu_mpi_req_list_new();
  1277. STARPU_PTHREAD_MUTEX_INIT(&detached_requests_mutex, NULL);
  1278. detached_requests = _starpu_mpi_req_list_new();
  1279. STARPU_PTHREAD_MUTEX_INIT(&mutex_posted_requests, NULL);
  1280. struct _starpu_mpi_argc_argv *argc_argv = malloc(sizeof(struct _starpu_mpi_argc_argv));
  1281. argc_argv->initialize_mpi = initialize_mpi;
  1282. argc_argv->argc = argc;
  1283. argc_argv->argv = argv;
  1284. #ifdef STARPU_MPI_ACTIVITY
  1285. hookid = starpu_progression_hook_register(progression_hook_func, NULL);
  1286. STARPU_ASSERT_MSG(hookid >= 0, "starpu_progression_hook_register failed");
  1287. #endif /* STARPU_MPI_ACTIVITY */
  1288. STARPU_PTHREAD_CREATE(&progress_thread, NULL, _starpu_mpi_progress_thread_func, argc_argv);
  1289. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  1290. while (!running)
  1291. STARPU_PTHREAD_COND_WAIT(&cond_progression, &mutex);
  1292. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  1293. return 0;
  1294. }
  1295. int starpu_mpi_init(int *argc, char ***argv, int initialize_mpi)
  1296. {
  1297. return _starpu_mpi_initialize(argc, argv, initialize_mpi);
  1298. }
  1299. int starpu_mpi_initialize(void)
  1300. {
  1301. return _starpu_mpi_initialize(NULL, NULL, 0);
  1302. }
  1303. int starpu_mpi_initialize_extended(int *rank, int *world_size)
  1304. {
  1305. int ret;
  1306. ret = _starpu_mpi_initialize(NULL, NULL, 1);
  1307. if (ret == 0)
  1308. {
  1309. _STARPU_DEBUG("Calling MPI_Comm_rank\n");
  1310. MPI_Comm_rank(MPI_COMM_WORLD, rank);
  1311. MPI_Comm_size(MPI_COMM_WORLD, world_size);
  1312. }
  1313. return ret;
  1314. }
  1315. int starpu_mpi_shutdown(void)
  1316. {
  1317. void *value;
  1318. int rank, world_size;
  1319. /* We need to get the rank before calling MPI_Finalize to pass to _starpu_mpi_comm_amounts_display() */
  1320. MPI_Comm_rank(MPI_COMM_WORLD, &rank);
  1321. MPI_Comm_size(MPI_COMM_WORLD, &world_size);
  1322. /* kill the progression thread */
  1323. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  1324. running = 0;
  1325. STARPU_PTHREAD_COND_BROADCAST(&cond_progression);
  1326. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  1327. starpu_pthread_join(progress_thread, &value);
  1328. #ifdef STARPU_MPI_ACTIVITY
  1329. starpu_progression_hook_deregister(hookid);
  1330. #endif /* STARPU_MPI_ACTIVITY */
  1331. TRACE_MPI_STOP(rank, world_size);
  1332. /* free the request queues */
  1333. _starpu_mpi_req_list_delete(detached_requests);
  1334. _starpu_mpi_req_list_delete(new_requests);
  1335. _starpu_mpi_comm_amounts_display(rank);
  1336. _starpu_mpi_comm_amounts_free();
  1337. _starpu_mpi_cache_free(world_size);
  1338. return 0;
  1339. }