copy_driver.c 35 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2010-2017 Université de Bordeaux
  4. * Copyright (C) 2010, 2011, 2013, 2016 CNRS
  5. * Copyright (C) 2016 INRIA
  6. *
  7. * StarPU is free software; you can redistribute it and/or modify
  8. * it under the terms of the GNU Lesser General Public License as published by
  9. * the Free Software Foundation; either version 2.1 of the License, or (at
  10. * your option) any later version.
  11. *
  12. * StarPU is distributed in the hope that it will be useful, but
  13. * WITHOUT ANY WARRANTY; without even the implied warranty of
  14. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  15. *
  16. * See the GNU Lesser General Public License in COPYING.LGPL for more details.
  17. */
  18. #include <starpu.h>
  19. #include <common/config.h>
  20. #include <common/utils.h>
  21. #include <core/sched_policy.h>
  22. #include <datawizard/datastats.h>
  23. #include <datawizard/memory_nodes.h>
  24. #include <drivers/disk/driver_disk.h>
  25. #include <drivers/mpi/driver_mpi_sink.h>
  26. #include <drivers/mpi/driver_mpi_source.h>
  27. #include <drivers/mpi/driver_mpi_common.h>
  28. #include <common/fxt.h>
  29. #include "copy_driver.h"
  30. #include "memalloc.h"
  31. #include <starpu_opencl.h>
  32. #include <starpu_cuda.h>
  33. #include <profiling/profiling.h>
  34. #include <core/disk.h>
  35. #ifdef STARPU_SIMGRID
  36. #include <core/simgrid.h>
  37. #endif
  38. void _starpu_wake_all_blocked_workers_on_node(unsigned nodeid)
  39. {
  40. /* wake up all workers on that memory node */
  41. unsigned cond_id;
  42. struct _starpu_memory_node_descr * const descr = _starpu_memory_node_get_description();
  43. starpu_pthread_mutex_t *mymutex = NULL;
  44. starpu_pthread_cond_t *mycond = NULL;
  45. const int myworkerid = starpu_worker_get_id();
  46. if (myworkerid >= 0)
  47. starpu_worker_get_sched_condition(myworkerid, &mymutex, &mycond);
  48. STARPU_PTHREAD_RWLOCK_RDLOCK(&descr->conditions_rwlock);
  49. unsigned nconds = descr->condition_count[nodeid];
  50. for (cond_id = 0; cond_id < nconds; cond_id++)
  51. {
  52. struct _starpu_cond_and_mutex *condition;
  53. condition = &descr->conditions_attached_to_node[nodeid][cond_id];
  54. if (condition->mutex == mymutex)
  55. /* No need to wake myself, and I might be called from
  56. * the scheduler with mutex locked, through
  57. * starpu_prefetch_task_input_on_node */
  58. continue;
  59. /* wake anybody waiting on that condition */
  60. STARPU_PTHREAD_MUTEX_LOCK_SCHED(condition->mutex);
  61. STARPU_PTHREAD_COND_BROADCAST(condition->cond);
  62. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(condition->mutex);
  63. }
  64. STARPU_PTHREAD_RWLOCK_UNLOCK(&descr->conditions_rwlock);
  65. #ifdef STARPU_SIMGRID
  66. starpu_pthread_queue_broadcast(&_starpu_simgrid_transfer_queue[nodeid]);
  67. #endif
  68. }
  69. void starpu_wake_all_blocked_workers(void)
  70. {
  71. /* workers may be blocked on the various queues' conditions */
  72. unsigned cond_id;
  73. struct _starpu_memory_node_descr * const descr = _starpu_memory_node_get_description();
  74. starpu_pthread_mutex_t *mymutex = NULL;
  75. starpu_pthread_cond_t *mycond = NULL;
  76. const int myworkerid = starpu_worker_get_id();
  77. if (myworkerid >= 0)
  78. starpu_worker_get_sched_condition(myworkerid, &mymutex, &mycond);
  79. STARPU_PTHREAD_RWLOCK_RDLOCK(&descr->conditions_rwlock);
  80. unsigned nconds = descr->total_condition_count;
  81. for (cond_id = 0; cond_id < nconds; cond_id++)
  82. {
  83. struct _starpu_cond_and_mutex *condition;
  84. condition = &descr->conditions_all[cond_id];
  85. if (condition->mutex == mymutex)
  86. /* No need to wake myself, and I might be called from
  87. * the scheduler with mutex locked, through
  88. * starpu_prefetch_task_input_on_node */
  89. continue;
  90. /* wake anybody waiting on that condition */
  91. STARPU_PTHREAD_MUTEX_LOCK_SCHED(condition->mutex);
  92. STARPU_PTHREAD_COND_BROADCAST(condition->cond);
  93. STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(condition->mutex);
  94. }
  95. STARPU_PTHREAD_RWLOCK_UNLOCK(&descr->conditions_rwlock);
  96. #ifdef STARPU_SIMGRID
  97. unsigned workerid, nodeid;
  98. for (workerid = 0; workerid < starpu_worker_get_count(); workerid++)
  99. starpu_pthread_queue_broadcast(&_starpu_simgrid_task_queue[workerid]);
  100. for (nodeid = 0; nodeid < starpu_memory_nodes_get_count(); nodeid++)
  101. starpu_pthread_queue_broadcast(&_starpu_simgrid_transfer_queue[nodeid]);
  102. #endif
  103. }
  104. #ifdef STARPU_USE_FXT
  105. /* we need to identify each communication so that we can match the beginning
  106. * and the end of a communication in the trace, so we use a unique identifier
  107. * per communication */
  108. static unsigned long communication_cnt = 0;
  109. #endif
  110. static int copy_data_1_to_1_generic(starpu_data_handle_t handle,
  111. struct _starpu_data_replicate *src_replicate,
  112. struct _starpu_data_replicate *dst_replicate,
  113. struct _starpu_data_request *req)
  114. {
  115. unsigned src_node = src_replicate->memory_node;
  116. unsigned dst_node = dst_replicate->memory_node;
  117. STARPU_ASSERT(src_replicate->refcnt);
  118. STARPU_ASSERT(dst_replicate->refcnt);
  119. STARPU_ASSERT(src_replicate->allocated);
  120. STARPU_ASSERT(dst_replicate->allocated);
  121. #ifdef STARPU_SIMGRID
  122. return _starpu_simgrid_transfer(handle->ops->get_size(handle), src_node, dst_node, req);
  123. #else /* !SIMGRID */
  124. int ret = 0;
  125. const struct starpu_data_copy_methods *copy_methods = handle->ops->copy_methods;
  126. enum starpu_node_kind src_kind = starpu_node_get_kind(src_node);
  127. enum starpu_node_kind dst_kind = starpu_node_get_kind(dst_node);
  128. #ifdef STARPU_USE_CUDA
  129. cudaError_t cures;
  130. cudaStream_t stream;
  131. #endif
  132. void *src_interface = src_replicate->data_interface;
  133. void *dst_interface = dst_replicate->data_interface;
  134. #if defined(STARPU_USE_CUDA) && defined(HAVE_CUDA_MEMCPY_PEER) && !defined(STARPU_SIMGRID)
  135. if ((src_kind == STARPU_CUDA_RAM) || (dst_kind == STARPU_CUDA_RAM))
  136. {
  137. unsigned devid;
  138. if ((src_kind == STARPU_CUDA_RAM) && (dst_kind == STARPU_CUDA_RAM))
  139. {
  140. /* GPU-GPU transfer, issue it from the destination */
  141. devid = _starpu_memory_node_get_devid(dst_node);
  142. }
  143. else
  144. {
  145. unsigned node = (dst_kind == STARPU_CUDA_RAM)?dst_node:src_node;
  146. devid = _starpu_memory_node_get_devid(node);
  147. }
  148. starpu_cuda_set_device(devid);
  149. }
  150. #endif
  151. switch (_STARPU_MEMORY_NODE_TUPLE(src_kind,dst_kind))
  152. {
  153. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_CPU_RAM):
  154. /* STARPU_CPU_RAM -> STARPU_CPU_RAM */
  155. if (copy_methods->ram_to_ram)
  156. copy_methods->ram_to_ram(src_interface, src_node, dst_interface, dst_node);
  157. else
  158. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, req ? &req->async_channel : NULL);
  159. break;
  160. #ifdef STARPU_USE_CUDA
  161. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CUDA_RAM,STARPU_CPU_RAM):
  162. /* only the proper CUBLAS thread can initiate this directly ! */
  163. #if !defined(HAVE_CUDA_MEMCPY_PEER)
  164. STARPU_ASSERT(_starpu_memory_node_get_local_key() == src_node);
  165. #endif
  166. if (!req || starpu_asynchronous_copy_disabled() || starpu_asynchronous_cuda_copy_disabled() ||
  167. !(copy_methods->cuda_to_ram_async || copy_methods->any_to_any))
  168. {
  169. /* this is not associated to a request so it's synchronous */
  170. STARPU_ASSERT(copy_methods->cuda_to_ram || copy_methods->any_to_any);
  171. if (copy_methods->cuda_to_ram)
  172. copy_methods->cuda_to_ram(src_interface, src_node, dst_interface, dst_node);
  173. else
  174. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  175. }
  176. else
  177. {
  178. req->async_channel.type = STARPU_CUDA_RAM;
  179. cures = cudaEventCreateWithFlags(&req->async_channel.event.cuda_event, cudaEventDisableTiming);
  180. if (STARPU_UNLIKELY(cures != cudaSuccess)) STARPU_CUDA_REPORT_ERROR(cures);
  181. stream = starpu_cuda_get_out_transfer_stream(src_node);
  182. if (copy_methods->cuda_to_ram_async)
  183. ret = copy_methods->cuda_to_ram_async(src_interface, src_node, dst_interface, dst_node, stream);
  184. else
  185. {
  186. STARPU_ASSERT(copy_methods->any_to_any);
  187. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  188. }
  189. cures = cudaEventRecord(req->async_channel.event.cuda_event, stream);
  190. if (STARPU_UNLIKELY(cures != cudaSuccess)) STARPU_CUDA_REPORT_ERROR(cures);
  191. }
  192. break;
  193. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_CUDA_RAM):
  194. /* STARPU_CPU_RAM -> CUBLAS_RAM */
  195. /* only the proper CUBLAS thread can initiate this ! */
  196. #if !defined(HAVE_CUDA_MEMCPY_PEER)
  197. STARPU_ASSERT(_starpu_memory_node_get_local_key() == dst_node);
  198. #endif
  199. if (!req || starpu_asynchronous_copy_disabled() || starpu_asynchronous_cuda_copy_disabled() ||
  200. !(copy_methods->ram_to_cuda_async || copy_methods->any_to_any))
  201. {
  202. /* this is not associated to a request so it's synchronous */
  203. STARPU_ASSERT(copy_methods->ram_to_cuda || copy_methods->any_to_any);
  204. if (copy_methods->ram_to_cuda)
  205. copy_methods->ram_to_cuda(src_interface, src_node, dst_interface, dst_node);
  206. else
  207. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  208. }
  209. else
  210. {
  211. req->async_channel.type = STARPU_CUDA_RAM;
  212. cures = cudaEventCreateWithFlags(&req->async_channel.event.cuda_event, cudaEventDisableTiming);
  213. if (STARPU_UNLIKELY(cures != cudaSuccess))
  214. STARPU_CUDA_REPORT_ERROR(cures);
  215. stream = starpu_cuda_get_in_transfer_stream(dst_node);
  216. if (copy_methods->ram_to_cuda_async)
  217. ret = copy_methods->ram_to_cuda_async(src_interface, src_node, dst_interface, dst_node, stream);
  218. else
  219. {
  220. STARPU_ASSERT(copy_methods->any_to_any);
  221. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  222. }
  223. cures = cudaEventRecord(req->async_channel.event.cuda_event, stream);
  224. if (STARPU_UNLIKELY(cures != cudaSuccess))
  225. STARPU_CUDA_REPORT_ERROR(cures);
  226. }
  227. break;
  228. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CUDA_RAM,STARPU_CUDA_RAM):
  229. /* CUDA - CUDA transfer */
  230. if (!req || starpu_asynchronous_copy_disabled() || starpu_asynchronous_cuda_copy_disabled() ||
  231. !(copy_methods->cuda_to_cuda_async || copy_methods->any_to_any))
  232. {
  233. STARPU_ASSERT(copy_methods->cuda_to_cuda || copy_methods->any_to_any);
  234. /* this is not associated to a request so it's synchronous */
  235. if (copy_methods->cuda_to_cuda)
  236. copy_methods->cuda_to_cuda(src_interface, src_node, dst_interface, dst_node);
  237. else
  238. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  239. }
  240. else
  241. {
  242. req->async_channel.type = STARPU_CUDA_RAM;
  243. cures = cudaEventCreateWithFlags(&req->async_channel.event.cuda_event, cudaEventDisableTiming);
  244. if (STARPU_UNLIKELY(cures != cudaSuccess)) STARPU_CUDA_REPORT_ERROR(cures);
  245. stream = starpu_cuda_get_peer_transfer_stream(src_node, dst_node);
  246. if (copy_methods->cuda_to_cuda_async)
  247. ret = copy_methods->cuda_to_cuda_async(src_interface, src_node, dst_interface, dst_node, stream);
  248. else
  249. {
  250. STARPU_ASSERT(copy_methods->any_to_any);
  251. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  252. }
  253. cures = cudaEventRecord(req->async_channel.event.cuda_event, stream);
  254. if (STARPU_UNLIKELY(cures != cudaSuccess)) STARPU_CUDA_REPORT_ERROR(cures);
  255. }
  256. break;
  257. #endif
  258. #ifdef STARPU_USE_OPENCL
  259. case _STARPU_MEMORY_NODE_TUPLE(STARPU_OPENCL_RAM,STARPU_CPU_RAM):
  260. /* OpenCL -> RAM */
  261. STARPU_ASSERT(_starpu_memory_node_get_local_key() == src_node);
  262. if (!req || starpu_asynchronous_copy_disabled() || starpu_asynchronous_opencl_copy_disabled() ||
  263. !(copy_methods->opencl_to_ram_async || copy_methods->any_to_any))
  264. {
  265. STARPU_ASSERT(copy_methods->opencl_to_ram || copy_methods->any_to_any);
  266. /* this is not associated to a request so it's synchronous */
  267. if (copy_methods->opencl_to_ram)
  268. copy_methods->opencl_to_ram(src_interface, src_node, dst_interface, dst_node);
  269. else
  270. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  271. }
  272. else
  273. {
  274. req->async_channel.type = STARPU_OPENCL_RAM;
  275. if (copy_methods->opencl_to_ram_async)
  276. ret = copy_methods->opencl_to_ram_async(src_interface, src_node, dst_interface, dst_node, &(req->async_channel.event.opencl_event));
  277. else
  278. {
  279. STARPU_ASSERT(copy_methods->any_to_any);
  280. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  281. }
  282. }
  283. break;
  284. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_OPENCL_RAM):
  285. /* STARPU_CPU_RAM -> STARPU_OPENCL_RAM */
  286. STARPU_ASSERT(_starpu_memory_node_get_local_key() == dst_node);
  287. if (!req || starpu_asynchronous_copy_disabled() || starpu_asynchronous_opencl_copy_disabled() ||
  288. !(copy_methods->ram_to_opencl_async || copy_methods->any_to_any))
  289. {
  290. STARPU_ASSERT(copy_methods->ram_to_opencl || copy_methods->any_to_any);
  291. /* this is not associated to a request so it's synchronous */
  292. if (copy_methods->ram_to_opencl)
  293. copy_methods->ram_to_opencl(src_interface, src_node, dst_interface, dst_node);
  294. else
  295. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  296. }
  297. else
  298. {
  299. req->async_channel.type = STARPU_OPENCL_RAM;
  300. if (copy_methods->ram_to_opencl_async)
  301. ret = copy_methods->ram_to_opencl_async(src_interface, src_node, dst_interface, dst_node, &(req->async_channel.event.opencl_event));
  302. else
  303. {
  304. STARPU_ASSERT(copy_methods->any_to_any);
  305. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  306. }
  307. }
  308. break;
  309. case _STARPU_MEMORY_NODE_TUPLE(STARPU_OPENCL_RAM,STARPU_OPENCL_RAM):
  310. /* STARPU_OPENCL_RAM -> STARPU_OPENCL_RAM */
  311. STARPU_ASSERT(_starpu_memory_node_get_local_key() == dst_node || _starpu_memory_node_get_local_key() == src_node);
  312. if (!req || starpu_asynchronous_copy_disabled() || starpu_asynchronous_opencl_copy_disabled() ||
  313. !(copy_methods->opencl_to_opencl_async || copy_methods->any_to_any))
  314. {
  315. STARPU_ASSERT(copy_methods->opencl_to_opencl || copy_methods->any_to_any);
  316. /* this is not associated to a request so it's synchronous */
  317. if (copy_methods->opencl_to_opencl)
  318. copy_methods->opencl_to_opencl(src_interface, src_node, dst_interface, dst_node);
  319. else
  320. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  321. }
  322. else
  323. {
  324. req->async_channel.type = STARPU_OPENCL_RAM;
  325. if (copy_methods->opencl_to_opencl_async)
  326. ret = copy_methods->opencl_to_opencl_async(src_interface, src_node, dst_interface, dst_node, &(req->async_channel.event.opencl_event));
  327. else
  328. {
  329. STARPU_ASSERT(copy_methods->any_to_any);
  330. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  331. }
  332. }
  333. break;
  334. #endif
  335. #ifdef STARPU_USE_MIC
  336. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_MIC_RAM):
  337. /* RAM -> MIC */
  338. if (!req || starpu_asynchronous_copy_disabled() || starpu_asynchronous_mic_copy_disabled() ||
  339. !(copy_methods->ram_to_mic_async || copy_methods->any_to_any))
  340. {
  341. /* this is not associated to a request so it's synchronous */
  342. STARPU_ASSERT(copy_methods->ram_to_mic || copy_methods->any_to_any);
  343. if (copy_methods->ram_to_mic)
  344. copy_methods->ram_to_mic(src_interface, src_node, dst_interface, dst_node);
  345. else
  346. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  347. }
  348. else
  349. {
  350. req->async_channel.type = STARPU_MIC_RAM;
  351. if (copy_methods->ram_to_mic_async)
  352. ret = copy_methods->ram_to_mic_async(src_interface, src_node, dst_interface, dst_node);
  353. else
  354. {
  355. STARPU_ASSERT(copy_methods->any_to_any);
  356. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  357. }
  358. _starpu_mic_init_event(&(req->async_channel.event.mic_event), dst_node);
  359. }
  360. break;
  361. case _STARPU_MEMORY_NODE_TUPLE(STARPU_MIC_RAM,STARPU_CPU_RAM):
  362. /* MIC -> RAM */
  363. if (!req || starpu_asynchronous_copy_disabled() || starpu_asynchronous_mic_copy_disabled() ||
  364. !(copy_methods->mic_to_ram_async || copy_methods->any_to_any))
  365. {
  366. /* this is not associated to a request so it's synchronous */
  367. STARPU_ASSERT(copy_methods->mic_to_ram || copy_methods->any_to_any);
  368. if (copy_methods->mic_to_ram)
  369. copy_methods->mic_to_ram(src_interface, src_node, dst_interface, dst_node);
  370. else
  371. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  372. }
  373. else
  374. {
  375. req->async_channel.type = STARPU_MIC_RAM;
  376. if (copy_methods->mic_to_ram_async)
  377. ret = copy_methods->mic_to_ram_async(src_interface, src_node, dst_interface, dst_node);
  378. else
  379. {
  380. STARPU_ASSERT(copy_methods->any_to_any);
  381. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  382. }
  383. _starpu_mic_init_event(&(req->async_channel.event.mic_event), src_node);
  384. }
  385. break;
  386. /* TODO: MIC -> MIC */
  387. #endif
  388. #ifdef STARPU_USE_MPI_MASTER_SLAVE
  389. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_MPI_MS_RAM):
  390. if (!req || starpu_asynchronous_copy_disabled() || starpu_asynchronous_mpi_ms_copy_disabled() ||
  391. !(copy_methods->ram_to_mpi_ms_async || copy_methods->any_to_any))
  392. {
  393. /* this is not associated to a request so it's synchronous */
  394. STARPU_ASSERT(copy_methods->ram_to_mpi_ms || copy_methods->any_to_any);
  395. if (copy_methods->ram_to_mpi_ms)
  396. copy_methods->ram_to_mpi_ms(src_interface, src_node, dst_interface, dst_node);
  397. else
  398. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  399. }
  400. else
  401. {
  402. req->async_channel.type = STARPU_MPI_MS_RAM;
  403. if(copy_methods->ram_to_mpi_ms_async)
  404. ret = copy_methods->ram_to_mpi_ms_async(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  405. else
  406. {
  407. STARPU_ASSERT(copy_methods->any_to_any);
  408. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  409. }
  410. }
  411. break;
  412. case _STARPU_MEMORY_NODE_TUPLE(STARPU_MPI_MS_RAM,STARPU_CPU_RAM):
  413. if (!req || starpu_asynchronous_copy_disabled() || starpu_asynchronous_mpi_ms_copy_disabled() ||
  414. !(copy_methods->mpi_ms_to_ram_async || copy_methods->any_to_any))
  415. {
  416. /* this is not associated to a request so it's synchronous */
  417. STARPU_ASSERT(copy_methods->mpi_ms_to_ram || copy_methods->any_to_any);
  418. if (copy_methods->mpi_ms_to_ram)
  419. copy_methods->mpi_ms_to_ram(src_interface, src_node, dst_interface, dst_node);
  420. else
  421. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  422. }
  423. else
  424. {
  425. req->async_channel.type = STARPU_MPI_MS_RAM;
  426. if(copy_methods->mpi_ms_to_ram_async)
  427. ret = copy_methods->mpi_ms_to_ram_async(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  428. else
  429. {
  430. STARPU_ASSERT(copy_methods->any_to_any);
  431. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  432. }
  433. }
  434. break;
  435. case _STARPU_MEMORY_NODE_TUPLE(STARPU_MPI_MS_RAM,STARPU_MPI_MS_RAM):
  436. if (!req || starpu_asynchronous_copy_disabled() || starpu_asynchronous_mpi_ms_copy_disabled() ||
  437. !(copy_methods->mpi_ms_to_mpi_ms_async || copy_methods->any_to_any))
  438. {
  439. /* this is not associated to a request so it's synchronous */
  440. STARPU_ASSERT(copy_methods->mpi_ms_to_mpi_ms || copy_methods->any_to_any);
  441. if (copy_methods->mpi_ms_to_mpi_ms)
  442. copy_methods->mpi_ms_to_mpi_ms(src_interface, src_node, dst_interface, dst_node);
  443. else
  444. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  445. }
  446. else
  447. {
  448. req->async_channel.type = STARPU_MPI_MS_RAM;
  449. if(copy_methods->mpi_ms_to_mpi_ms_async)
  450. ret = copy_methods->mpi_ms_to_mpi_ms_async(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  451. else
  452. {
  453. STARPU_ASSERT(copy_methods->any_to_any);
  454. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, &req->async_channel);
  455. }
  456. }
  457. break;
  458. #endif
  459. #ifdef STARPU_USE_SCC
  460. /* SCC RAM associated to the master process is considered as
  461. * the main memory node. */
  462. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_SCC_RAM):
  463. /* master private SCC RAM -> slave private SCC RAM */
  464. if (copy_methods->scc_src_to_sink)
  465. copy_methods->scc_src_to_sink(src_interface, src_node, dst_interface, dst_node);
  466. else
  467. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  468. break;
  469. case _STARPU_MEMORY_NODE_TUPLE(STARPU_SCC_RAM,STARPU_CPU_RAM):
  470. /* slave private SCC RAM -> master private SCC RAM */
  471. if (copy_methods->scc_sink_to_src)
  472. copy_methods->scc_sink_to_src(src_interface, src_node, dst_interface, dst_node);
  473. else
  474. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  475. break;
  476. case _STARPU_MEMORY_NODE_TUPLE(STARPU_SCC_RAM,STARPU_SCC_RAM):
  477. /* slave private SCC RAM -> slave private SCC RAM */
  478. if (copy_methods->scc_sink_to_sink)
  479. copy_methods->scc_sink_to_sink(src_interface, src_node, dst_interface, dst_node);
  480. else
  481. copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, NULL);
  482. break;
  483. #endif
  484. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_DISK_RAM):
  485. if(copy_methods->any_to_any)
  486. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, req && !starpu_asynchronous_copy_disabled() ? &req->async_channel : NULL);
  487. else
  488. {
  489. void *obj = starpu_data_handle_to_pointer(handle, dst_node);
  490. void * ptr = NULL;
  491. starpu_ssize_t size = 0;
  492. handle->ops->pack_data(handle, src_node, &ptr, &size);
  493. ret = _starpu_disk_full_write(src_node, dst_node, obj, ptr, size, req && !starpu_asynchronous_copy_disabled() ? &req->async_channel : NULL);
  494. if (ret == 0)
  495. /* write is already finished, ptr was allocated in pack_data */
  496. starpu_free_flags(ptr, size, 0);
  497. /* For now, asynchronous is not supported */
  498. STARPU_ASSERT(ret == 0);
  499. }
  500. break;
  501. case _STARPU_MEMORY_NODE_TUPLE(STARPU_DISK_RAM,STARPU_CPU_RAM):
  502. if(copy_methods->any_to_any)
  503. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, req && !starpu_asynchronous_copy_disabled() ? &req->async_channel : NULL);
  504. else
  505. {
  506. void *obj = starpu_data_handle_to_pointer(handle, src_node);
  507. void * ptr = NULL;
  508. size_t size = 0;
  509. ret = _starpu_disk_full_read(src_node, dst_node, obj, &ptr, &size, req && !starpu_asynchronous_copy_disabled() ? &req->async_channel : NULL);
  510. if (ret == 0)
  511. {
  512. /* read is already finished, we can already unpack */
  513. handle->ops->unpack_data(handle, dst_node, ptr, size);
  514. /* ptr is allocated in full_read */
  515. starpu_free_flags(ptr, size, 0);
  516. }
  517. /* For now, asynchronous is not supported */
  518. STARPU_ASSERT(ret == 0);
  519. }
  520. break;
  521. case _STARPU_MEMORY_NODE_TUPLE(STARPU_DISK_RAM,STARPU_DISK_RAM):
  522. ret = copy_methods->any_to_any(src_interface, src_node, dst_interface, dst_node, req ? &req->async_channel : NULL);
  523. break;
  524. default:
  525. STARPU_ABORT();
  526. break;
  527. }
  528. return ret;
  529. #endif /* !SIMGRID */
  530. }
  531. int STARPU_ATTRIBUTE_WARN_UNUSED_RESULT _starpu_driver_copy_data_1_to_1(starpu_data_handle_t handle,
  532. struct _starpu_data_replicate *src_replicate,
  533. struct _starpu_data_replicate *dst_replicate,
  534. unsigned donotread,
  535. struct _starpu_data_request *req,
  536. unsigned may_alloc,
  537. unsigned prefetch STARPU_ATTRIBUTE_UNUSED)
  538. {
  539. if (!donotread)
  540. {
  541. STARPU_ASSERT(src_replicate->allocated);
  542. STARPU_ASSERT(src_replicate->refcnt);
  543. }
  544. unsigned src_node = src_replicate->memory_node;
  545. unsigned dst_node = dst_replicate->memory_node;
  546. /* first make sure the destination has an allocated buffer */
  547. if (!dst_replicate->allocated)
  548. {
  549. if (!may_alloc || _starpu_is_reclaiming(dst_node))
  550. /* We're not supposed to allocate there at the moment */
  551. return -ENOMEM;
  552. int ret_alloc = _starpu_allocate_memory_on_node(handle, dst_replicate, req ? req->prefetch : 0);
  553. if (ret_alloc)
  554. return -ENOMEM;
  555. }
  556. STARPU_ASSERT(dst_replicate->allocated);
  557. STARPU_ASSERT(dst_replicate->refcnt);
  558. /* if there is no need to actually read the data,
  559. * we do not perform any transfer */
  560. if (!donotread)
  561. {
  562. unsigned long STARPU_ATTRIBUTE_UNUSED com_id = 0;
  563. size_t size = _starpu_data_get_size(handle);
  564. _starpu_bus_update_profiling_info((int)src_node, (int)dst_node, size);
  565. #ifdef STARPU_USE_FXT
  566. com_id = STARPU_ATOMIC_ADDL(&communication_cnt, 1);
  567. if (req)
  568. req->com_id = com_id;
  569. #endif
  570. dst_replicate->initialized = 1;
  571. _STARPU_TRACE_START_DRIVER_COPY(src_node, dst_node, size, com_id, prefetch, handle);
  572. int ret_copy = copy_data_1_to_1_generic(handle, src_replicate, dst_replicate, req);
  573. if (!req)
  574. /* Synchronous, this is already finished */
  575. _STARPU_TRACE_END_DRIVER_COPY(src_node, dst_node, size, com_id, prefetch);
  576. return ret_copy;
  577. }
  578. return 0;
  579. }
  580. /* This can be used by interfaces to easily transfer a piece of data without
  581. * caring about the particular transfer methods. */
  582. /* This should either return 0 if the transfer is complete, or -EAGAIN if the
  583. * transfer is still pending, and will have to be waited for by
  584. * _starpu_driver_test_request_completion/_starpu_driver_wait_request_completion
  585. */
  586. int starpu_interface_copy(uintptr_t src, size_t src_offset, unsigned src_node, uintptr_t dst, size_t dst_offset, unsigned dst_node, size_t size, void *async_data)
  587. {
  588. struct _starpu_async_channel *async_channel = async_data;
  589. enum starpu_node_kind src_kind = starpu_node_get_kind(src_node);
  590. enum starpu_node_kind dst_kind = starpu_node_get_kind(dst_node);
  591. switch (_STARPU_MEMORY_NODE_TUPLE(src_kind,dst_kind))
  592. {
  593. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_CPU_RAM):
  594. memcpy((void *) (dst + dst_offset), (void *) (src + src_offset), size);
  595. return 0;
  596. #ifdef STARPU_USE_CUDA
  597. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CUDA_RAM,STARPU_CPU_RAM):
  598. return starpu_cuda_copy_async_sync(
  599. (void*) (src + src_offset), src_node,
  600. (void*) (dst + dst_offset), dst_node,
  601. size,
  602. async_channel?starpu_cuda_get_out_transfer_stream(src_node):NULL,
  603. cudaMemcpyDeviceToHost);
  604. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_CUDA_RAM):
  605. return starpu_cuda_copy_async_sync(
  606. (void*) (src + src_offset), src_node,
  607. (void*) (dst + dst_offset), dst_node,
  608. size,
  609. async_channel?starpu_cuda_get_in_transfer_stream(dst_node):NULL,
  610. cudaMemcpyHostToDevice);
  611. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CUDA_RAM,STARPU_CUDA_RAM):
  612. return starpu_cuda_copy_async_sync(
  613. (void*) (src + src_offset), src_node,
  614. (void*) (dst + dst_offset), dst_node,
  615. size,
  616. async_channel?starpu_cuda_get_peer_transfer_stream(src_node, dst_node):NULL,
  617. cudaMemcpyDeviceToDevice);
  618. #endif
  619. #ifdef STARPU_USE_OPENCL
  620. case _STARPU_MEMORY_NODE_TUPLE(STARPU_OPENCL_RAM,STARPU_CPU_RAM):
  621. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_OPENCL_RAM):
  622. case _STARPU_MEMORY_NODE_TUPLE(STARPU_OPENCL_RAM,STARPU_OPENCL_RAM):
  623. return starpu_opencl_copy_async_sync(
  624. src, src_offset, src_node,
  625. dst, dst_offset, dst_node,
  626. size,
  627. &async_channel->event.opencl_event);
  628. #endif
  629. #ifdef STARPU_USE_MIC
  630. case _STARPU_MEMORY_NODE_TUPLE(STARPU_MIC_RAM,STARPU_CPU_RAM):
  631. if (async_data)
  632. return _starpu_mic_copy_mic_to_ram_async(
  633. (void*) (src + src_offset), src_node,
  634. (void*) (dst + dst_offset), dst_node,
  635. size);
  636. else
  637. return _starpu_mic_copy_mic_to_ram(
  638. (void*) (src + src_offset), src_node,
  639. (void*) (dst + dst_offset), dst_node,
  640. size);
  641. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_MIC_RAM):
  642. if (async_data)
  643. return _starpu_mic_copy_ram_to_mic_async(
  644. (void*) (src + src_offset), src_node,
  645. (void*) (dst + dst_offset), dst_node,
  646. size);
  647. else
  648. return _starpu_mic_copy_ram_to_mic(
  649. (void*) (src + src_offset), src_node,
  650. (void*) (dst + dst_offset), dst_node,
  651. size);
  652. /* TODO: MIC->MIC */
  653. #endif
  654. #ifdef STARPU_USE_SCC
  655. case _STARPU_MEMORY_NODE_TUPLE(STARPU_SCC_RAM,STARPU_CPU_RAM):
  656. return _starpu_scc_copy_sink_to_src(
  657. (void*) (src + src_offset), src_node,
  658. (void*) (dst + dst_offset), dst_node,
  659. size);
  660. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM,STARPU_SCC_RAM):
  661. return _starpu_scc_copy_src_to_sink(
  662. (void*) (src + src_offset), src_node,
  663. (void*) (dst + dst_offset), dst_node,
  664. size);
  665. case _STARPU_MEMORY_NODE_TUPLE(STARPU_SCC_RAM,STARPU_SCC_RAM):
  666. return _starpu_scc_copy_sink_to_sink(
  667. (void*) (src + src_offset), src_node,
  668. (void*) (dst + dst_offset), dst_node,
  669. size);
  670. #endif
  671. #ifdef STARPU_USE_MPI_MASTER_SLAVE
  672. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM, STARPU_MPI_MS_RAM):
  673. if (async_data)
  674. return _starpu_mpi_copy_ram_to_mpi_async(
  675. (void*) (src + src_offset), src_node,
  676. (void*) (dst + dst_offset), dst_node,
  677. size, async_data);
  678. else
  679. return _starpu_mpi_copy_ram_to_mpi_sync(
  680. (void*) (src + src_offset), src_node,
  681. (void*) (dst + dst_offset), dst_node,
  682. size);
  683. case _STARPU_MEMORY_NODE_TUPLE(STARPU_MPI_MS_RAM, STARPU_CPU_RAM):
  684. if (async_data)
  685. return _starpu_mpi_copy_mpi_to_ram_async(
  686. (void*) (src + src_offset), src_node,
  687. (void*) (dst + dst_offset), dst_node,
  688. size, async_data);
  689. else
  690. return _starpu_mpi_copy_mpi_to_ram_sync(
  691. (void*) (src + src_offset), src_node,
  692. (void*) (dst + dst_offset), dst_node,
  693. size);
  694. case _STARPU_MEMORY_NODE_TUPLE(STARPU_MPI_MS_RAM, STARPU_MPI_MS_RAM):
  695. if (async_data)
  696. return _starpu_mpi_copy_sink_to_sink_async(
  697. (void*) (src + src_offset), src_node,
  698. (void*) (dst + dst_offset), dst_node,
  699. size, async_data);
  700. else
  701. return _starpu_mpi_copy_sink_to_sink_sync(
  702. (void*) (src + src_offset), src_node,
  703. (void*) (dst + dst_offset), dst_node,
  704. size);
  705. #endif
  706. case _STARPU_MEMORY_NODE_TUPLE(STARPU_CPU_RAM, STARPU_DISK_RAM):
  707. {
  708. return _starpu_disk_copy_src_to_disk(
  709. (void*) (src + src_offset), src_node,
  710. (void*) dst, dst_offset, dst_node,
  711. size, async_channel);
  712. }
  713. case _STARPU_MEMORY_NODE_TUPLE(STARPU_DISK_RAM, STARPU_CPU_RAM):
  714. return _starpu_disk_copy_disk_to_src(
  715. (void*) src, src_offset, src_node,
  716. (void*) (dst + dst_offset), dst_node,
  717. size, async_channel);
  718. case _STARPU_MEMORY_NODE_TUPLE(STARPU_DISK_RAM, STARPU_DISK_RAM):
  719. return _starpu_disk_copy_disk_to_disk(
  720. (void*) src, src_offset, src_node,
  721. (void*) dst, dst_offset, dst_node,
  722. size, async_channel);
  723. default:
  724. STARPU_ABORT();
  725. return -1;
  726. }
  727. return 0;
  728. }
  729. void _starpu_driver_wait_request_completion(struct _starpu_async_channel *async_channel)
  730. {
  731. #ifdef STARPU_SIMGRID
  732. _starpu_simgrid_wait_transfer_event(&async_channel->event);
  733. #else /* !SIMGRID */
  734. enum starpu_node_kind kind = async_channel->type;
  735. #ifdef STARPU_USE_CUDA
  736. cudaEvent_t event;
  737. cudaError_t cures;
  738. #endif
  739. switch (kind)
  740. {
  741. #ifdef STARPU_USE_CUDA
  742. case STARPU_CUDA_RAM:
  743. event = (*async_channel).event.cuda_event;
  744. cures = cudaEventSynchronize(event);
  745. if (STARPU_UNLIKELY(cures))
  746. STARPU_CUDA_REPORT_ERROR(cures);
  747. cures = cudaEventDestroy(event);
  748. if (STARPU_UNLIKELY(cures))
  749. STARPU_CUDA_REPORT_ERROR(cures);
  750. break;
  751. #endif
  752. #ifdef STARPU_USE_OPENCL
  753. case STARPU_OPENCL_RAM:
  754. {
  755. cl_int err;
  756. if ((*async_channel).event.opencl_event == NULL)
  757. STARPU_ABORT();
  758. err = clWaitForEvents(1, &((*async_channel).event.opencl_event));
  759. if (STARPU_UNLIKELY(err != CL_SUCCESS))
  760. STARPU_OPENCL_REPORT_ERROR(err);
  761. err = clReleaseEvent((*async_channel).event.opencl_event);
  762. if (STARPU_UNLIKELY(err != CL_SUCCESS))
  763. STARPU_OPENCL_REPORT_ERROR(err);
  764. break;
  765. }
  766. #endif
  767. #ifdef STARPU_USE_MIC
  768. case STARPU_MIC_RAM:
  769. _starpu_mic_wait_request_completion(&(async_channel->event.mic_event));
  770. break;
  771. #endif
  772. #ifdef STARPU_USE_MPI_MASTER_SLAVE
  773. case STARPU_MPI_MS_RAM:
  774. _starpu_mpi_common_wait_event(async_channel);
  775. break;
  776. #endif
  777. case STARPU_CPU_RAM:
  778. starpu_disk_wait_request(async_channel);
  779. break;
  780. default:
  781. STARPU_ABORT();
  782. }
  783. #endif /* !SIMGRID */
  784. }
  785. unsigned _starpu_driver_test_request_completion(struct _starpu_async_channel *async_channel)
  786. {
  787. #ifdef STARPU_SIMGRID
  788. return _starpu_simgrid_test_transfer_event(&async_channel->event);
  789. #else /* !SIMGRID */
  790. enum starpu_node_kind kind = async_channel->type;
  791. unsigned success = 0;
  792. #ifdef STARPU_USE_CUDA
  793. cudaEvent_t event;
  794. #endif
  795. switch (kind)
  796. {
  797. #ifdef STARPU_USE_CUDA
  798. case STARPU_CUDA_RAM:
  799. event = (*async_channel).event.cuda_event;
  800. cudaError_t cures = cudaEventQuery(event);
  801. success = (cures == cudaSuccess);
  802. if (success)
  803. cudaEventDestroy(event);
  804. else if (cures != cudaErrorNotReady)
  805. STARPU_CUDA_REPORT_ERROR(cures);
  806. break;
  807. #endif
  808. #ifdef STARPU_USE_OPENCL
  809. case STARPU_OPENCL_RAM:
  810. {
  811. cl_int event_status;
  812. cl_event opencl_event = (*async_channel).event.opencl_event;
  813. if (opencl_event == NULL) STARPU_ABORT();
  814. cl_int err = clGetEventInfo(opencl_event, CL_EVENT_COMMAND_EXECUTION_STATUS, sizeof(event_status), &event_status, NULL);
  815. if (STARPU_UNLIKELY(err != CL_SUCCESS))
  816. STARPU_OPENCL_REPORT_ERROR(err);
  817. if (event_status < 0)
  818. STARPU_OPENCL_REPORT_ERROR(event_status);
  819. if (event_status == CL_COMPLETE)
  820. {
  821. err = clReleaseEvent(opencl_event);
  822. if (STARPU_UNLIKELY(err != CL_SUCCESS)) STARPU_OPENCL_REPORT_ERROR(err);
  823. }
  824. success = (event_status == CL_COMPLETE);
  825. break;
  826. }
  827. #endif
  828. #ifdef STARPU_USE_MIC
  829. case STARPU_MIC_RAM:
  830. success = _starpu_mic_request_is_complete(&(async_channel->event.mic_event));
  831. break;
  832. #endif
  833. #ifdef STARPU_USE_MPI_MASTER_SLAVE
  834. case STARPU_MPI_MS_RAM:
  835. success = _starpu_mpi_common_test_event(async_channel);
  836. break;
  837. #endif
  838. case STARPU_DISK_RAM:
  839. success = starpu_disk_test_request(async_channel);
  840. break;
  841. case STARPU_CPU_RAM:
  842. default:
  843. STARPU_ABORT();
  844. }
  845. return success;
  846. #endif /* !SIMGRID */
  847. }