ring_async.c 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2009, 2010, 2014-2017 Université de Bordeaux
  4. * Copyright (C) 2010, 2011, 2012, 2013, 2016, 2017 CNRS
  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 <starpu_mpi.h>
  18. #include "helper.h"
  19. #ifdef STARPU_QUICK_CHECK
  20. # define NITER 32
  21. #elif !defined(STARPU_LONG_CHECK)
  22. # define NITER 256
  23. #else
  24. # define NITER 2048
  25. #endif
  26. #ifdef STARPU_USE_CUDA
  27. extern void increment_cuda(void *descr[], STARPU_ATTRIBUTE_UNUSED void *_args);
  28. #endif
  29. void increment_cpu(void *descr[], STARPU_ATTRIBUTE_UNUSED void *_args)
  30. {
  31. int *tokenptr = (int *)STARPU_VECTOR_GET_PTR(descr[0]);
  32. (*tokenptr)++;
  33. }
  34. /* Dummy cost function for simgrid */
  35. static double cost_function(struct starpu_task *task STARPU_ATTRIBUTE_UNUSED, unsigned nimpl STARPU_ATTRIBUTE_UNUSED)
  36. {
  37. return 0.000001;
  38. }
  39. static struct starpu_perfmodel dumb_model =
  40. {
  41. .type = STARPU_COMMON,
  42. .cost_function = cost_function
  43. };
  44. static struct starpu_codelet increment_cl =
  45. {
  46. #ifdef STARPU_USE_CUDA
  47. .cuda_funcs = {increment_cuda},
  48. #endif
  49. .cpu_funcs = {increment_cpu},
  50. .nbuffers = 1,
  51. .modes = {STARPU_RW},
  52. .model = &dumb_model
  53. };
  54. void increment_token(starpu_data_handle_t token_handle)
  55. {
  56. struct starpu_task *task = starpu_task_create();
  57. task->cl = &increment_cl;
  58. task->handles[0] = token_handle;
  59. task->synchronous = 1;
  60. int ret = starpu_task_submit(task);
  61. STARPU_CHECK_RETURN_VALUE(ret, "starpu_task_submit");
  62. }
  63. int main(int argc, char **argv)
  64. {
  65. int ret, rank, size;
  66. int mpi_init;
  67. int token = 42;
  68. starpu_data_handle_t token_handle;
  69. MPI_INIT_THREAD(&argc, &argv, MPI_THREAD_SERIALIZED, &mpi_init);
  70. ret = starpu_init(NULL);
  71. STARPU_CHECK_RETURN_VALUE(ret, "starpu_init");
  72. ret = starpu_mpi_init(&argc, &argv, mpi_init);
  73. STARPU_CHECK_RETURN_VALUE(ret, "starpu_mpi_init");
  74. starpu_mpi_comm_rank(MPI_COMM_WORLD, &rank);
  75. starpu_mpi_comm_size(MPI_COMM_WORLD, &size);
  76. if (size < 2 || (starpu_cpu_worker_get_count() + starpu_cuda_worker_get_count() == 0))
  77. {
  78. if (rank == 0)
  79. {
  80. if (size < 2)
  81. FPRINTF(stderr, "We need at least 2 processes.\n");
  82. else
  83. FPRINTF(stderr, "We need at least 1 CPU or CUDA worker.\n");
  84. }
  85. starpu_mpi_shutdown();
  86. starpu_shutdown();
  87. if (!mpi_init)
  88. MPI_Finalize();
  89. return STARPU_TEST_SKIPPED;
  90. }
  91. starpu_vector_data_register(&token_handle, STARPU_MAIN_RAM, (uintptr_t)&token, 1, sizeof(token));
  92. int nloops = NITER;
  93. int loop;
  94. int last_loop = nloops - 1;
  95. int last_rank = size - 1;
  96. for (loop = 0; loop < nloops; loop++)
  97. {
  98. int tag = loop*size + rank;
  99. if (loop == 0 && rank == 0)
  100. {
  101. starpu_data_acquire(token_handle, STARPU_W);
  102. token = 0;
  103. FPRINTF(stdout, "Start with token value %d\n", token);
  104. starpu_data_release(token_handle);
  105. }
  106. else
  107. {
  108. MPI_Status status;
  109. starpu_mpi_req req;
  110. starpu_mpi_irecv(token_handle, &req, (rank+size-1)%size, tag, MPI_COMM_WORLD);
  111. starpu_mpi_wait(&req, &status);
  112. }
  113. increment_token(token_handle);
  114. if (loop == last_loop && rank == last_rank)
  115. {
  116. starpu_data_acquire(token_handle, STARPU_R);
  117. FPRINTF(stdout, "Finished : token value %d\n", token);
  118. starpu_data_release(token_handle);
  119. }
  120. else
  121. {
  122. starpu_mpi_req req;
  123. MPI_Status status;
  124. starpu_mpi_isend(token_handle, &req, (rank+1)%size, tag+1, MPI_COMM_WORLD);
  125. starpu_mpi_wait(&req, &status);
  126. }
  127. }
  128. starpu_data_unregister(token_handle);
  129. starpu_mpi_shutdown();
  130. starpu_shutdown();
  131. if (!mpi_init)
  132. MPI_Finalize();
  133. #ifndef STARPU_SIMGRID
  134. if (rank == last_rank)
  135. {
  136. FPRINTF(stderr, "[%d] token = %d == %d * %d ?\n", rank, token, nloops, size);
  137. STARPU_ASSERT(token == nloops*size);
  138. }
  139. #endif
  140. return 0;
  141. }