sync_with_data_with_mem_non_blocking.c 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2010 Université de Bordeaux 1
  4. * Copyright (C) 2010, 2011 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 <stdio.h>
  18. #include <unistd.h>
  19. #include <errno.h>
  20. #include <starpu.h>
  21. #include <stdlib.h>
  22. #include <pthread.h>
  23. #include "../common/helper.h"
  24. #define NBUFFERS 64
  25. #define NITER 128
  26. #define VECTORSIZE 1024
  27. float *buffer[NBUFFERS];
  28. starpu_data_handle v_handle[NBUFFERS];
  29. static void dummy_codelet(void *descr[], __attribute__ ((unused)) void *_args)
  30. {
  31. }
  32. static starpu_codelet cl = {
  33. .where = STARPU_CPU|STARPU_CUDA|STARPU_OPENCL,
  34. .cpu_func = dummy_codelet,
  35. #ifdef STARPU_USE_CUDA
  36. .cuda_func = dummy_codelet,
  37. #endif
  38. #ifdef STARPU_USE_OPENCL
  39. .opencl_func = dummy_codelet,
  40. #endif
  41. .nbuffers = 1
  42. };
  43. int use_handle(starpu_data_handle handle)
  44. {
  45. int ret;
  46. struct starpu_task *task;
  47. task = starpu_task_create();
  48. task->cl = &cl;
  49. task->buffers[0].handle = handle;
  50. task->buffers[0].mode = STARPU_RW;
  51. task->detach = 0;
  52. ret = starpu_task_submit(task);
  53. return ret;
  54. }
  55. static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  56. static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
  57. static unsigned n_synced_buffers;
  58. void callback_sync_data(void *arg __attribute__ ((unused)))
  59. {
  60. PTHREAD_MUTEX_LOCK(&mutex);
  61. n_synced_buffers++;
  62. if (n_synced_buffers == NBUFFERS)
  63. PTHREAD_COND_SIGNAL(&cond);
  64. PTHREAD_MUTEX_UNLOCK(&mutex);
  65. }
  66. int main(int argc, char **argv)
  67. {
  68. int ret;
  69. ret = starpu_init(NULL);
  70. STARPU_CHECK_RETURN_VALUE(ret, "starpu_init");
  71. /* Allocate all buffers and register them to StarPU */
  72. unsigned b;
  73. for (b = 0; b < NBUFFERS; b++)
  74. {
  75. ret = starpu_malloc((void **)&buffer[b], VECTORSIZE);
  76. STARPU_CHECK_RETURN_VALUE(ret, "starpu_malloc");
  77. starpu_vector_data_register(&v_handle[b], 0,
  78. (uintptr_t)buffer[b], VECTORSIZE, sizeof(char));
  79. starpu_data_set_sequential_consistency_flag(v_handle[b], 0);
  80. }
  81. unsigned iter;
  82. for (iter = 0; iter < NITER; iter++)
  83. {
  84. /* Use the buffers on the different workers so that it may not
  85. * be in main memory anymore */
  86. for (b = 0; b < NBUFFERS; b++) {
  87. ret = use_handle(v_handle[b]);
  88. if (ret == -ENODEV) goto enodev;
  89. STARPU_CHECK_RETURN_VALUE(ret, "starpu_task_submit");
  90. }
  91. ret = starpu_task_wait_for_all();
  92. STARPU_CHECK_RETURN_VALUE(ret, "starpu_task_wait_for_all");
  93. PTHREAD_MUTEX_LOCK(&mutex);
  94. n_synced_buffers = 0;
  95. PTHREAD_MUTEX_UNLOCK(&mutex);
  96. /* Grab the different pieces of data into main memory */
  97. for (b = 0; b < NBUFFERS; b++)
  98. {
  99. ret = starpu_data_acquire_cb(v_handle[b], STARPU_RW,
  100. callback_sync_data, NULL);
  101. STARPU_CHECK_RETURN_VALUE(ret, "starpu_data_acquire_cb");
  102. }
  103. /* Wait for all buffers to be available */
  104. PTHREAD_MUTEX_LOCK(&mutex);
  105. while (n_synced_buffers != NBUFFERS)
  106. PTHREAD_COND_WAIT(&cond, &mutex);
  107. PTHREAD_MUTEX_UNLOCK(&mutex);
  108. /* Release them */
  109. for (b = 0; b < NBUFFERS; b++)
  110. starpu_data_release(v_handle[b]);
  111. }
  112. /* do some cleanup */
  113. for (b = 0; b < NBUFFERS; b++) {
  114. starpu_data_unregister(v_handle[b]);
  115. starpu_free(buffer[b]);
  116. }
  117. starpu_shutdown();
  118. return 0;
  119. enodev:
  120. fprintf(stderr, "WARNING: No one can execute this task\n");
  121. /* yes, we do not perform the computation but we did detect that no one
  122. * could perform the kernel, so this is not an error from StarPU */
  123. starpu_shutdown();
  124. return 77;
  125. }