pxlu.c 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800
  1. /*
  2. * StarPU
  3. * Copyright (C) INRIA 2008-2010 (see AUTHORS file)
  4. *
  5. * This program 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. * This program 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 "pxlu.h"
  17. #include "pxlu_kernels.h"
  18. #define MPI_TAG11(k) ((1U << 16) | (k))
  19. #define MPI_TAG12(k, j) ((2U << 16) | (j)<<8 | (k))
  20. #define MPI_TAG21(k, i) ((3U << 16) | (i)<<8 | (k))
  21. #define TAG11(k) ((starpu_tag_t)( (1ULL<<50) | (unsigned long long)(k)))
  22. #define TAG12(k,i) ((starpu_tag_t)(((2ULL<<50) | (((unsigned long long)(k))<<32) \
  23. | (unsigned long long)(i))))
  24. #define TAG21(k,j) ((starpu_tag_t)(((3ULL<<50) | (((unsigned long long)(k))<<32) \
  25. | (unsigned long long)(j))))
  26. #define TAG22(k,i,j) ((starpu_tag_t)(((4ULL<<50) | ((unsigned long long)(k)<<32) \
  27. | ((unsigned long long)(i)<<16) \
  28. | (unsigned long long)(j))))
  29. #define TAG11_SAVE(k) ((starpu_tag_t)( (5ULL<<50) | (unsigned long long)(k)))
  30. #define TAG12_SAVE(k,i) ((starpu_tag_t)(((6ULL<<50) | (((unsigned long long)(k))<<32) \
  31. | (unsigned long long)(i))))
  32. #define TAG21_SAVE(k,j) ((starpu_tag_t)(((7ULL<<50) | (((unsigned long long)(k))<<32) \
  33. | (unsigned long long)(j))))
  34. #define TAG11_SAVE_PARTIAL(k) ((starpu_tag_t)( (8ULL<<50) | (unsigned long long)(k)))
  35. #define TAG12_SAVE_PARTIAL(k,i) ((starpu_tag_t)(((9ULL<<50) | (((unsigned long long)(k))<<32) \
  36. | (unsigned long long)(i))))
  37. #define TAG21_SAVE_PARTIAL(k,j) ((starpu_tag_t)(((10ULL<<50) | (((unsigned long long)(k))<<32) \
  38. | (unsigned long long)(j))))
  39. #define STARPU_TAG_INIT ((starpu_tag_t)(11ULL<<50))
  40. //#define VERBOSE_INIT 1
  41. #define DEBUG 1
  42. static unsigned no_prio = 0;
  43. static unsigned nblocks = 0;
  44. static int rank = -1;
  45. static int world_size = -1;
  46. struct callback_arg {
  47. unsigned i, j, k;
  48. };
  49. /*
  50. * Various
  51. */
  52. static struct debug_info *create_debug_info(unsigned i, unsigned j, unsigned k)
  53. {
  54. struct debug_info *info = malloc(sizeof(struct debug_info));
  55. info->i = i;
  56. info->j = j;
  57. info->k = k;
  58. return info;
  59. }
  60. static struct starpu_task *create_task(starpu_tag_t id)
  61. {
  62. struct starpu_task *task = starpu_task_create();
  63. task->cl_arg = NULL;
  64. task->use_tag = 1;
  65. task->tag_id = id;
  66. return task;
  67. }
  68. /* Send handle to every node appearing in the mask, and unlock tag once the
  69. * transfers are done. */
  70. static void send_data_to_mask(starpu_data_handle handle, int *rank_mask, int mpi_tag, starpu_tag_t tag)
  71. {
  72. unsigned cnt = 0;
  73. STARPU_ASSERT(handle != STARPU_POISON_PTR);
  74. int rank_array[world_size];
  75. int comm_array[world_size];
  76. int mpi_tag_array[world_size];
  77. starpu_data_handle handle_array[world_size];
  78. unsigned r;
  79. for (r = 0; r < world_size; r++)
  80. {
  81. if (rank_mask[r]) {
  82. rank_array[cnt] = r;
  83. comm_array[cnt] = MPI_COMM_WORLD;
  84. mpi_tag_array[cnt] = mpi_tag;
  85. handle_array[cnt] = handle;
  86. cnt++;
  87. }
  88. }
  89. if (cnt == 0)
  90. {
  91. /* In case there is no message to send, we release the tag at
  92. * once */
  93. starpu_tag_notify_from_apps(tag);
  94. }
  95. else {
  96. starpu_mpi_isend_array_detached_unlock_tag(cnt, handle_array,
  97. rank_array, mpi_tag_array, comm_array, tag);
  98. }
  99. }
  100. /* Initiate a receive request once all dependencies are fulfilled and unlock
  101. * tag 'unlocked_tag' once it's done. */
  102. struct recv_when_done_callback_arg {
  103. int source;
  104. int mpi_tag;
  105. starpu_data_handle handle;
  106. starpu_tag_t unlocked_tag;
  107. };
  108. static void callback_receive_when_done(void *_arg)
  109. {
  110. struct recv_when_done_callback_arg *arg = _arg;
  111. starpu_mpi_irecv_detached_unlock_tag(arg->handle, arg->source,
  112. arg->mpi_tag, MPI_COMM_WORLD, arg->unlocked_tag);
  113. free(arg);
  114. }
  115. static void receive_when_deps_are_done(unsigned ndeps, starpu_tag_t *deps_tags,
  116. int source, int mpi_tag,
  117. starpu_data_handle handle,
  118. starpu_tag_t partial_tag,
  119. starpu_tag_t unlocked_tag)
  120. {
  121. STARPU_ASSERT(handle != STARPU_POISON_PTR);
  122. struct recv_when_done_callback_arg *arg =
  123. malloc(sizeof(struct recv_when_done_callback_arg));
  124. arg->source = source;
  125. arg->mpi_tag = mpi_tag;
  126. arg->handle = handle;
  127. arg->unlocked_tag = unlocked_tag;
  128. if (ndeps == 0)
  129. {
  130. callback_receive_when_done(arg);
  131. return;
  132. }
  133. starpu_create_sync_task(partial_tag, ndeps, deps_tags,
  134. callback_receive_when_done, arg);
  135. }
  136. /*
  137. * Task 11 (diagonal factorization)
  138. */
  139. static void create_task_11_recv(unsigned k)
  140. {
  141. unsigned i, j;
  142. /* The current node is not computing that task, so we receive the block
  143. * with MPI */
  144. /* We don't issue a MPI receive request until everyone using the
  145. * temporary buffer is done : 11_(k-1) can be used by 12_(k-1)j and
  146. * 21(k-1)i with i,j >= k */
  147. unsigned ndeps = 0;
  148. starpu_tag_t tag_array[2*nblocks];
  149. if (k > 0)
  150. for (i = k; i < nblocks; i++)
  151. {
  152. if (rank == get_block_rank(i, k))
  153. tag_array[ndeps++] = TAG21(k-1, i);
  154. }
  155. if (k > 0)
  156. for (j = k; j < nblocks; j++)
  157. {
  158. if (rank == get_block_rank(k, j))
  159. tag_array[ndeps++] = TAG12(k-1, j);
  160. }
  161. int source = get_block_rank(k, k);
  162. starpu_data_handle block_handle = STARPU_PLU(get_tmp_11_block_handle)();
  163. int mpi_tag = MPI_TAG11(k);
  164. starpu_tag_t partial_tag = TAG11_SAVE_PARTIAL(k);
  165. starpu_tag_t unlocked_tag = TAG11_SAVE(k);
  166. // fprintf(stderr, "NODE %d - 11 (%d) - recv when done ndeps %d - tag array %lx\n", rank, k, ndeps, tag_array[0]);
  167. receive_when_deps_are_done(ndeps, tag_array, source, mpi_tag, block_handle, partial_tag, unlocked_tag);
  168. }
  169. static void find_nodes_using_11(unsigned k, int *rank_mask)
  170. {
  171. memset(rank_mask, 0, world_size*sizeof(int));
  172. /* Block 11_k is used to compute 12_kj + 12ki with i,j > k */
  173. unsigned i;
  174. for (i = k; i < nblocks; i++)
  175. {
  176. int r = get_block_rank(i, k);
  177. rank_mask[r] = 1;
  178. }
  179. unsigned j;
  180. for (j = k; j < nblocks; j++)
  181. {
  182. int r = get_block_rank(k, j);
  183. rank_mask[r] = 1;
  184. }
  185. }
  186. static void callback_task_11_real(void *_arg)
  187. {
  188. struct callback_arg *arg = _arg;
  189. unsigned k = arg->k;
  190. /* Find all the nodes potentially requiring this block */
  191. int rank_mask[world_size];
  192. find_nodes_using_11(k, rank_mask);
  193. rank_mask[rank] = 0;
  194. /* Send the block to those nodes */
  195. starpu_data_handle block_handle = STARPU_PLU(get_block_handle)(k, k);
  196. starpu_tag_t tag = TAG11_SAVE(k);
  197. int mpi_tag = MPI_TAG11(k);
  198. send_data_to_mask(block_handle, rank_mask, mpi_tag, tag);
  199. free(arg);
  200. }
  201. static void create_task_11_real(unsigned k)
  202. {
  203. struct starpu_task *task = create_task(TAG11(k));
  204. task->cl = &STARPU_PLU(cl11);
  205. task->cl_arg = create_debug_info(k, k, k);
  206. /* which sub-data is manipulated ? */
  207. task->buffers[0].handle = STARPU_PLU(get_block_handle)(k, k);
  208. task->buffers[0].mode = STARPU_RW;
  209. struct callback_arg *arg = malloc(sizeof(struct callback_arg));
  210. arg->k = k;
  211. task->callback_func = callback_task_11_real;
  212. task->callback_arg = arg;
  213. /* this is an important task */
  214. if (!no_prio)
  215. task->priority = MAX_PRIO;
  216. /* enforce dependencies ... */
  217. if (k > 0) {
  218. starpu_tag_declare_deps(TAG11(k), 1, TAG22(k-1, k, k));
  219. }
  220. else {
  221. starpu_tag_declare_deps(TAG11(k), 1, STARPU_TAG_INIT);
  222. }
  223. starpu_submit_task(task);
  224. }
  225. static void create_task_11(unsigned k)
  226. {
  227. if (get_block_rank(k, k) == rank)
  228. {
  229. #ifdef VERBOSE_INIT
  230. fprintf(stderr, "CREATE real task 11(%d) (TAG11_SAVE(%d) = %lx) on node %d\n", k, k, TAG11_SAVE(k), rank);
  231. #endif
  232. create_task_11_real(k);
  233. }
  234. else {
  235. /* We don't handle the task, but perhaps we have to generate MPI transfers. */
  236. int rank_mask[world_size];
  237. find_nodes_using_11(k, rank_mask);
  238. if (rank_mask[rank])
  239. {
  240. #ifdef VERBOSE_INIT
  241. fprintf(stderr, "create RECV task 11(%d) on node %d\n", k, rank);
  242. #endif
  243. create_task_11_recv(k);
  244. }
  245. else {
  246. #ifdef VERBOSE_INIT
  247. fprintf(stderr, "Node %d needs not 11(%d)\n", rank, k);
  248. #endif
  249. }
  250. }
  251. }
  252. /*
  253. * Task 12 (Update lower left (TRSM))
  254. */
  255. static void create_task_12_recv(unsigned k, unsigned j)
  256. {
  257. unsigned i;
  258. /* The current node is not computing that task, so we receive the block
  259. * with MPI */
  260. /* We don't issue a MPI receive request until everyone using the
  261. * temporary buffer is done : 12_(k-1)j can be used by 22_(k-1)ij with
  262. * i >= k */
  263. unsigned ndeps = 0;
  264. starpu_tag_t tag_array[nblocks];
  265. if (k > 0)
  266. for (i = k; i < nblocks; i++)
  267. {
  268. if (rank == get_block_rank(i, j))
  269. tag_array[ndeps++] = TAG22(k-1, i, j);
  270. }
  271. int source = get_block_rank(k, j);
  272. starpu_data_handle block_handle = STARPU_PLU(get_tmp_12_block_handle)(j);
  273. int mpi_tag = MPI_TAG12(k, j);
  274. starpu_tag_t partial_tag = TAG12_SAVE_PARTIAL(k, j);
  275. starpu_tag_t unlocked_tag = TAG12_SAVE(k, j);
  276. receive_when_deps_are_done(ndeps, tag_array, source, mpi_tag, block_handle, partial_tag, unlocked_tag);
  277. }
  278. static void find_nodes_using_12(unsigned k, unsigned j, int *rank_mask)
  279. {
  280. memset(rank_mask, 0, world_size*sizeof(int));
  281. /* Block 12_kj is used to compute 22_kij with i > k */
  282. unsigned i;
  283. for (i = k; i < nblocks; i++)
  284. {
  285. int r = get_block_rank(i, j);
  286. rank_mask[r] = 1;
  287. }
  288. }
  289. static void callback_task_12_real(void *_arg)
  290. {
  291. struct callback_arg *arg = _arg;
  292. unsigned k = arg->k;
  293. unsigned j = arg->j;
  294. /* Find all the nodes potentially requiring this block */
  295. int rank_mask[world_size];
  296. find_nodes_using_12(k, j, rank_mask);
  297. rank_mask[rank] = 0;
  298. /* Send the block to those nodes */
  299. // starpu_data_handle block_handle = STARPU_PLU(get_block_handle)(j, k);
  300. starpu_data_handle block_handle = STARPU_PLU(get_block_handle)(k, j);
  301. starpu_tag_t tag = TAG12_SAVE(k, j);
  302. int mpi_tag = MPI_TAG12(k, j);
  303. send_data_to_mask(block_handle, rank_mask, mpi_tag, tag);
  304. free(arg);
  305. }
  306. static void create_task_12_real(unsigned k, unsigned j)
  307. {
  308. struct starpu_task *task = create_task(TAG12(k, j));
  309. #warning temporary fix :/
  310. // task->cl = &STARPU_PLU(cl12);
  311. task->cl = &STARPU_PLU(cl21);
  312. task->cl_arg = create_debug_info(j, j, k);
  313. int myrank;
  314. MPI_Comm_rank(MPI_COMM_WORLD, &myrank);
  315. STARPU_ASSERT(myrank == rank);
  316. /* which sub-data is manipulated ? */
  317. starpu_data_handle diag_block;
  318. if (get_block_rank(k, k) == rank)
  319. diag_block = STARPU_PLU(get_block_handle)(k, k);
  320. else
  321. diag_block = STARPU_PLU(get_tmp_11_block_handle)();
  322. task->buffers[0].handle = diag_block;
  323. task->buffers[0].mode = STARPU_R;
  324. task->buffers[1].handle = STARPU_PLU(get_block_handle)(k, j);
  325. task->buffers[1].mode = STARPU_RW;
  326. STARPU_ASSERT(get_block_rank(k, j) == rank);
  327. STARPU_ASSERT(STARPU_PLU(get_tmp_11_block_handle)() != STARPU_POISON_PTR);
  328. STARPU_ASSERT(task->buffers[0].handle != STARPU_POISON_PTR);
  329. STARPU_ASSERT(task->buffers[1].handle != STARPU_POISON_PTR);
  330. struct callback_arg *arg = malloc(sizeof(struct callback_arg));
  331. arg->j = j;
  332. arg->k = k;
  333. task->callback_func = callback_task_12_real;
  334. task->callback_arg = arg;
  335. if (!no_prio && (j == k+1)) {
  336. task->priority = MAX_PRIO;
  337. }
  338. /* enforce dependencies ... */
  339. if (k > 0) {
  340. starpu_tag_declare_deps(TAG12(k, j), 2, TAG11_SAVE(k), TAG22(k-1, k, j));
  341. }
  342. else {
  343. starpu_tag_declare_deps(TAG12(k, j), 1, TAG11_SAVE(k));
  344. }
  345. starpu_submit_task(task);
  346. }
  347. static void create_task_12(unsigned k, unsigned j)
  348. {
  349. if (get_block_rank(k, j) == rank)
  350. {
  351. #ifdef VERBOSE_INIT
  352. fprintf(stderr, "CREATE real task 12(k = %d, j = %d) on node %d\n", k, j, rank);
  353. #endif
  354. create_task_12_real(k, j);
  355. }
  356. else {
  357. /* We don't handle the task, but perhaps we have to generate MPI transfers. */
  358. int rank_mask[world_size];
  359. find_nodes_using_12(k, j, rank_mask);
  360. if (rank_mask[rank])
  361. {
  362. #ifdef VERBOSE_INIT
  363. fprintf(stderr, "create RECV task 12(k = %d, j = %d) on node %d\n", k, j, rank);
  364. #endif
  365. create_task_12_recv(k, j);
  366. }
  367. else {
  368. #ifdef VERBOSE_INIT
  369. fprintf(stderr, "Node %d needs not 12(k=%d, i=%d)\n", rank, k, j);
  370. #endif
  371. }
  372. }
  373. }
  374. /*
  375. * Task 21 (Update upper right (TRSM))
  376. */
  377. static void create_task_21_recv(unsigned k, unsigned i)
  378. {
  379. unsigned j;
  380. /* The current node is not computing that task, so we receive the block
  381. * with MPI */
  382. /* We don't issue a MPI receive request until everyone using the
  383. * temporary buffer is done : 21_(k-1)i can be used by 22_(k-1)ij with
  384. * j >= k */
  385. unsigned ndeps = 0;
  386. starpu_tag_t tag_array[nblocks];
  387. if (k > 0)
  388. for (j = k; j < nblocks; j++)
  389. {
  390. if (rank == get_block_rank(i, j))
  391. tag_array[ndeps++] = TAG22(k-1, i, j);
  392. }
  393. int source = get_block_rank(i, k);
  394. starpu_data_handle block_handle = STARPU_PLU(get_tmp_21_block_handle)(i);
  395. int mpi_tag = MPI_TAG21(k, i);
  396. starpu_tag_t partial_tag = TAG21_SAVE_PARTIAL(k, i);
  397. starpu_tag_t unlocked_tag = TAG21_SAVE(k, i);
  398. // fprintf(stderr, "NODE %d - 21 (%d, %d) - recv when done ndeps %d - tag array %lx\n", rank, k, i, ndeps, tag_array[0]);
  399. receive_when_deps_are_done(ndeps, tag_array, source, mpi_tag, block_handle, partial_tag, unlocked_tag);
  400. }
  401. static void find_nodes_using_21(unsigned k, unsigned i, int *rank_mask)
  402. {
  403. memset(rank_mask, 0, world_size*sizeof(int));
  404. /* Block 21_ki is used to compute 22_kij with j > k */
  405. unsigned j;
  406. for (j = k; j < nblocks; j++)
  407. {
  408. int r = get_block_rank(i, j);
  409. rank_mask[r] = 1;
  410. }
  411. }
  412. static void callback_task_21_real(void *_arg)
  413. {
  414. struct callback_arg *arg = _arg;
  415. unsigned k = arg->k;
  416. unsigned i = arg->i;
  417. /* Find all the nodes potentially requiring this block */
  418. int rank_mask[world_size];
  419. find_nodes_using_21(k, i, rank_mask);
  420. rank_mask[rank] = 0;
  421. /* Send the block to those nodes */
  422. starpu_data_handle block_handle = STARPU_PLU(get_block_handle)(i, k);
  423. starpu_tag_t tag = TAG21_SAVE(k, i);
  424. int mpi_tag = MPI_TAG21(k, i);
  425. send_data_to_mask(block_handle, rank_mask, mpi_tag, tag);
  426. free(arg);
  427. }
  428. static void create_task_21_real(unsigned k, unsigned i)
  429. {
  430. struct starpu_task *task = create_task(TAG21(k, i));
  431. #warning temporary fix
  432. // task->cl = &STARPU_PLU(cl21);
  433. task->cl = &STARPU_PLU(cl12);
  434. task->cl_arg = create_debug_info(i, i, k);
  435. /* which sub-data is manipulated ? */
  436. starpu_data_handle diag_block;
  437. if (get_block_rank(k, k) == rank)
  438. diag_block = STARPU_PLU(get_block_handle)(k, k);
  439. else
  440. diag_block = STARPU_PLU(get_tmp_11_block_handle)();
  441. task->buffers[0].handle = diag_block;
  442. task->buffers[0].mode = STARPU_R;
  443. task->buffers[1].handle = STARPU_PLU(get_block_handle)(i, k);
  444. task->buffers[1].mode = STARPU_RW;
  445. STARPU_ASSERT(STARPU_PLU(get_tmp_11_block_handle)() != STARPU_POISON_PTR);
  446. STARPU_ASSERT(task->buffers[0].handle != STARPU_POISON_PTR);
  447. STARPU_ASSERT(task->buffers[1].handle != STARPU_POISON_PTR);
  448. struct callback_arg *arg = malloc(sizeof(struct callback_arg));
  449. arg->i = i;
  450. arg->k = k;
  451. task->callback_func = callback_task_21_real;
  452. task->callback_arg = arg;
  453. if (!no_prio && (i == k+1)) {
  454. task->priority = MAX_PRIO;
  455. }
  456. /* enforce dependencies ... */
  457. if (k > 0) {
  458. starpu_tag_declare_deps(TAG21(k, i), 2, TAG11_SAVE(k), TAG22(k-1, i, k));
  459. }
  460. else {
  461. starpu_tag_declare_deps(TAG21(k, i), 1, TAG11_SAVE(k));
  462. }
  463. starpu_submit_task(task);
  464. }
  465. static void create_task_21(unsigned k, unsigned i)
  466. {
  467. if (get_block_rank(i, k) == rank)
  468. {
  469. #ifdef VERBOSE_INIT
  470. fprintf(stderr, "CREATE real task 21(k = %d, i = %d) on node %d\n", k, i, rank);
  471. #endif
  472. create_task_21_real(k, i);
  473. }
  474. else {
  475. /* We don't handle the task, but perhaps we have to generate MPI transfers. */
  476. int rank_mask[world_size];
  477. find_nodes_using_21(k, i, rank_mask);
  478. if (rank_mask[rank])
  479. {
  480. #ifdef VERBOSE_INIT
  481. fprintf(stderr, "create RECV task 21(k = %d, i = %d) on node %d\n", k, i, rank);
  482. #endif
  483. create_task_21_recv(k, i);
  484. }
  485. else {
  486. #ifdef VERBOSE_INIT
  487. fprintf(stderr, "Node %d needs not 21(k=%d, i=%d)\n", rank, k,i);
  488. #endif
  489. }
  490. }
  491. }
  492. /*
  493. * Task 22 (GEMM)
  494. */
  495. static void create_task_22_real(unsigned k, unsigned i, unsigned j)
  496. {
  497. // printf("task 22 k,i,j = %d,%d,%d TAG = %llx\n", k,i,j, TAG22(k,i,j));
  498. struct starpu_task *task = create_task(TAG22(k, i, j));
  499. task->cl = &STARPU_PLU(cl22);
  500. /* which sub-data is manipulated ? */
  501. /* produced by TAG21_SAVE(k, i) */
  502. starpu_data_handle block21;
  503. if (get_block_rank(i, k) == rank)
  504. {
  505. block21 = STARPU_PLU(get_block_handle)(i, k);
  506. }
  507. else
  508. block21 = STARPU_PLU(get_tmp_21_block_handle)(i);
  509. /* produced by TAG12_SAVE(k, j) */
  510. starpu_data_handle block12;
  511. if (get_block_rank(k, j) == rank)
  512. {
  513. // block12 = STARPU_PLU(get_block_handle)(j, k);
  514. block12 = STARPU_PLU(get_block_handle)(k, j);
  515. }
  516. else
  517. block12 = STARPU_PLU(get_tmp_12_block_handle)(j);
  518. //task->buffers[0].handle = block21;
  519. task->buffers[0].handle = block12;
  520. task->buffers[0].mode = STARPU_R;
  521. //task->buffers[1].handle = block12;
  522. task->buffers[1].handle = block21;
  523. task->buffers[1].mode = STARPU_R;
  524. /* produced by TAG22(k-1, i, j) */
  525. task->buffers[2].handle = STARPU_PLU(get_block_handle)(i, j);
  526. task->buffers[2].mode = STARPU_RW;
  527. STARPU_ASSERT(task->buffers[0].handle != STARPU_POISON_PTR);
  528. STARPU_ASSERT(task->buffers[1].handle != STARPU_POISON_PTR);
  529. STARPU_ASSERT(task->buffers[2].handle != STARPU_POISON_PTR);
  530. if (!no_prio && (i == k + 1) && (j == k +1) ) {
  531. task->priority = MAX_PRIO;
  532. }
  533. /* enforce dependencies ... */
  534. if (k > 0) {
  535. starpu_tag_declare_deps(TAG22(k, i, j), 3, TAG22(k-1, i, j), TAG12_SAVE(k, j), TAG21_SAVE(k, i));
  536. }
  537. else {
  538. starpu_tag_declare_deps(TAG22(k, i, j), 2, TAG12_SAVE(k, j), TAG21_SAVE(k, i));
  539. }
  540. starpu_submit_task(task);
  541. }
  542. static void create_task_22(unsigned k, unsigned i, unsigned j)
  543. {
  544. if (get_block_rank(i, j) == rank)
  545. {
  546. // fprintf(stderr, "CREATE real task 22(k = %d, i = %d, j = %d) on node %d\n", k, i, j, rank);
  547. create_task_22_real(k, i, j);
  548. }
  549. // else {
  550. // fprintf(stderr, "Node %d needs not 22(k=%d, i=%d, j = %d)\n", rank, k,i,j);
  551. // }
  552. }
  553. static void wait_tag_and_fetch_handle(starpu_tag_t tag, starpu_data_handle handle)
  554. {
  555. STARPU_ASSERT(handle != STARPU_POISON_PTR);
  556. starpu_tag_wait(tag);
  557. // fprintf(stderr, "Rank %d : tag %lx is done\n", rank, tag);
  558. starpu_sync_data_with_mem(handle, STARPU_R);
  559. // starpu_delete_data(handle);
  560. }
  561. static void wait_termination(void)
  562. {
  563. starpu_wait_all_tasks();
  564. unsigned k, i, j;
  565. for (k = 0; k < nblocks; k++)
  566. {
  567. /* Wait task 11k if needed */
  568. if (get_block_rank(k, k) == rank)
  569. {
  570. starpu_data_handle diag_block = STARPU_PLU(get_block_handle)(k, k);
  571. wait_tag_and_fetch_handle(TAG11_SAVE(k), diag_block);
  572. }
  573. for (i = k + 1; i < nblocks; i++)
  574. {
  575. /* Wait task 21ki is needed */
  576. if (get_block_rank(i, k) == rank)
  577. {
  578. starpu_data_handle block21 = STARPU_PLU(get_block_handle)(i, k);
  579. //starpu_data_handle block21 = STARPU_PLU(get_block_handle)(k, i);
  580. //fprintf(stderr, "BLOCK21 i %d k %d -> handle %p\n", i, k, block21);
  581. wait_tag_and_fetch_handle(TAG21_SAVE(k, i), block21);
  582. }
  583. }
  584. for (j = k + 1; j < nblocks; j++)
  585. {
  586. /* Wait task 12kj is needed */
  587. if (get_block_rank(k, j) == rank)
  588. {
  589. //starpu_data_handle block12 = STARPU_PLU(get_block_handle)(j, k);
  590. starpu_data_handle block12 = STARPU_PLU(get_block_handle)(k, j);
  591. //fprintf(stderr, "BLOCK12 j %d k %d -> handle %p\n", j, k, block12);
  592. wait_tag_and_fetch_handle(TAG12_SAVE(k, j), block12);
  593. }
  594. }
  595. }
  596. }
  597. /*
  598. * code to bootstrap the factorization
  599. */
  600. double STARPU_PLU(plu_main)(unsigned _nblocks, int _rank, int _world_size)
  601. {
  602. struct timeval start;
  603. struct timeval end;
  604. nblocks = _nblocks;
  605. rank = _rank;
  606. world_size = _world_size;
  607. struct starpu_task *entry_task = NULL;
  608. /* create all the DAG nodes */
  609. unsigned i,j,k;
  610. for (k = 0; k < nblocks; k++)
  611. {
  612. create_task_11(k);
  613. for (i = k+1; i<nblocks; i++)
  614. {
  615. create_task_12(k, i);
  616. create_task_21(k, i);
  617. }
  618. for (i = k+1; i<nblocks; i++)
  619. {
  620. for (j = k+1; j<nblocks; j++)
  621. {
  622. create_task_22(k, i, j);
  623. }
  624. }
  625. }
  626. int barrier_ret = MPI_Barrier(MPI_COMM_WORLD);
  627. STARPU_ASSERT(barrier_ret == MPI_SUCCESS);
  628. /* schedule the codelet */
  629. gettimeofday(&start, NULL);
  630. starpu_tag_notify_from_apps(STARPU_TAG_INIT);
  631. wait_termination();
  632. gettimeofday(&end, NULL);
  633. double timing = (double)((end.tv_sec - start.tv_sec)*1000000 + (end.tv_usec - start.tv_usec));
  634. // fprintf(stderr, "RANK %d -> took %lf ms\n", rank, timing/1000);
  635. return timing;
  636. }