coherency.c 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2009-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 <common/config.h>
  18. #include <datawizard/coherency.h>
  19. #include <datawizard/copy_driver.h>
  20. #include <datawizard/write_back.h>
  21. #include <core/dependencies/data_concurrency.h>
  22. #include <profiling/profiling.h>
  23. #include <math.h>
  24. #include <core/task.h>
  25. static int link_supports_direct_transfers(starpu_data_handle_t handle, unsigned src_node, unsigned dst_node, unsigned *handling_node);
  26. unsigned _starpu_select_src_node(starpu_data_handle_t handle, unsigned destination)
  27. {
  28. int src_node = -1;
  29. unsigned i;
  30. unsigned nnodes = starpu_memory_nodes_get_count();
  31. /* first find a valid copy, either a STARPU_OWNER or a STARPU_SHARED */
  32. unsigned node;
  33. unsigned src_node_mask = 0;
  34. size_t size = _starpu_data_get_size(handle);
  35. double cost = INFINITY;
  36. for (node = 0; node < nnodes; node++)
  37. {
  38. if (handle->per_node[node].state != STARPU_INVALID)
  39. {
  40. /* we found a copy ! */
  41. src_node_mask |= (1<<node);
  42. }
  43. }
  44. /* we should have found at least one copy ! */
  45. STARPU_ASSERT(src_node_mask != 0);
  46. /* Without knowing the size, we won't know the cost */
  47. if (!size)
  48. cost = 0;
  49. /* Check whether we have transfer cost for all nodes, if so, take the minimum */
  50. if (cost)
  51. for (i = 0; i < nnodes; i++)
  52. {
  53. if (src_node_mask & (1<<i))
  54. {
  55. double time = _starpu_predict_transfer_time(i, destination, size);
  56. unsigned handling_node;
  57. /* Avoid indirect transfers */
  58. if (!link_supports_direct_transfers(handle, i, destination, &handling_node))
  59. continue;
  60. if (_STARPU_IS_ZERO(time))
  61. {
  62. /* No estimation, will have to revert to dumb strategy */
  63. cost = 0.0;
  64. break;
  65. }
  66. else if (time < cost)
  67. {
  68. cost = time;
  69. src_node = i;
  70. }
  71. }
  72. }
  73. if (cost && src_node != -1)
  74. /* Could estimate through cost, return that */
  75. return src_node;
  76. /* Revert to dumb strategy: take RAM unless only a GPU has it */
  77. for (i = 0; i < nnodes; i++)
  78. {
  79. if (src_node_mask & (1<<i))
  80. {
  81. /* this is a potential candidate */
  82. src_node = i;
  83. /* however GPU are expensive sources, really !
  84. * Unless peer transfer is supported.
  85. * Other should be ok */
  86. if (
  87. #ifndef HAVE_CUDA_MEMCPY_PEER
  88. starpu_node_get_kind(i) != STARPU_CUDA_RAM &&
  89. #endif
  90. starpu_node_get_kind(i) != STARPU_OPENCL_RAM)
  91. break ;
  92. }
  93. }
  94. STARPU_ASSERT(src_node != -1);
  95. return src_node;
  96. }
  97. /* this may be called once the data is fetched with header and STARPU_RW-lock hold */
  98. void _starpu_update_data_state(starpu_data_handle_t handle,
  99. struct _starpu_data_replicate *requesting_replicate,
  100. enum starpu_access_mode mode)
  101. {
  102. /* There is nothing to do for relaxed coherency modes (scratch or
  103. * reductions) */
  104. if (!(mode & STARPU_RW))
  105. return;
  106. unsigned nnodes = starpu_memory_nodes_get_count();
  107. /* the data is present now */
  108. unsigned requesting_node = requesting_replicate->memory_node;
  109. requesting_replicate->requested[requesting_node] = 0;
  110. if (mode & STARPU_W)
  111. {
  112. /* the requesting node now has the only valid copy */
  113. unsigned node;
  114. for (node = 0; node < nnodes; node++)
  115. handle->per_node[node].state = STARPU_INVALID;
  116. requesting_replicate->state = STARPU_OWNER;
  117. }
  118. else
  119. { /* read only */
  120. if (requesting_replicate->state != STARPU_OWNER)
  121. {
  122. /* there was at least another copy of the data */
  123. unsigned node;
  124. for (node = 0; node < nnodes; node++)
  125. {
  126. struct _starpu_data_replicate *replicate = &handle->per_node[node];
  127. if (replicate->state != STARPU_INVALID)
  128. replicate->state = STARPU_SHARED;
  129. }
  130. requesting_replicate->state = STARPU_SHARED;
  131. }
  132. }
  133. }
  134. static int worker_supports_direct_access(unsigned node, unsigned handling_node)
  135. {
  136. if (node == handling_node)
  137. return 1;
  138. if (!_starpu_memory_node_get_nworkers(handling_node))
  139. /* No worker to process the request from that node */
  140. return 0;
  141. int type = starpu_node_get_kind(node);
  142. switch (type)
  143. {
  144. case STARPU_CUDA_RAM:
  145. #ifdef HAVE_CUDA_MEMCPY_PEER
  146. {
  147. enum starpu_node_kind kind = starpu_node_get_kind(handling_node);
  148. /* GPUs not always allow direct remote access: if CUDA4
  149. * is enabled, we allow two CUDA devices to communicate. */
  150. return kind == STARPU_CPU_RAM || kind == STARPU_CUDA_RAM;
  151. }
  152. #else
  153. /* Direct GPU-GPU transfers are not allowed in general */
  154. return 0;
  155. #endif
  156. case STARPU_OPENCL_RAM:
  157. return 0;
  158. default:
  159. return 1;
  160. }
  161. }
  162. static int link_supports_direct_transfers(starpu_data_handle_t handle, unsigned src_node, unsigned dst_node, unsigned *handling_node)
  163. {
  164. (void) handle; // unused
  165. /* XXX That's a hack until we get cudaMemcpy3DPeerAsync to work !
  166. * Perhaps not all data interface provide a direct GPU-GPU transfer
  167. * method ! */
  168. #if defined(STARPU_USE_CUDA) || defined(STARPU_SIMGRID)
  169. if (src_node != dst_node && starpu_node_get_kind(src_node) == STARPU_CUDA_RAM && starpu_node_get_kind(dst_node) == STARPU_CUDA_RAM)
  170. {
  171. const struct starpu_data_copy_methods *copy_methods = handle->ops->copy_methods;
  172. if (!copy_methods->cuda_to_cuda_async)
  173. return 0;
  174. }
  175. #endif
  176. /* Note: with CUDA, performance seems a bit better when issuing the transfer from the destination (tested without GPUDirect, but GPUDirect probably behave the same) */
  177. if (worker_supports_direct_access(src_node, dst_node))
  178. {
  179. *handling_node = dst_node;
  180. return 1;
  181. }
  182. if (worker_supports_direct_access(dst_node, src_node))
  183. {
  184. *handling_node = src_node;
  185. return 1;
  186. }
  187. return 0;
  188. }
  189. /* Determines the path of a request : each hop is defined by (src,dst) and the
  190. * node that handles the hop. The returned value indicates the number of hops,
  191. * and the max_len is the maximum number of hops (ie. the size of the
  192. * src_nodes, dst_nodes and handling_nodes arrays. */
  193. static int determine_request_path(starpu_data_handle_t handle,
  194. unsigned src_node, unsigned dst_node,
  195. enum starpu_access_mode mode, int max_len,
  196. unsigned *src_nodes, unsigned *dst_nodes,
  197. unsigned *handling_nodes)
  198. {
  199. if (!(mode & STARPU_R))
  200. {
  201. /* The destination node should only allocate the data, no transfer is required */
  202. STARPU_ASSERT(max_len >= 1);
  203. src_nodes[0] = 0; // ignored
  204. dst_nodes[0] = dst_node;
  205. handling_nodes[0] = dst_node;
  206. return 1;
  207. }
  208. unsigned handling_node;
  209. int link_is_valid = link_supports_direct_transfers(handle, src_node, dst_node, &handling_node);
  210. if (!link_is_valid)
  211. {
  212. /* We need an intermediate hop to implement data staging
  213. * through main memory. */
  214. STARPU_ASSERT(max_len >= 2);
  215. /* XXX we hardcode 0 as the RAM node ... */
  216. /* GPU -> RAM */
  217. src_nodes[0] = src_node;
  218. dst_nodes[0] = 0;
  219. handling_nodes[0] = src_node;
  220. /* RAM -> GPU */
  221. src_nodes[1] = 0;
  222. dst_nodes[1] = dst_node;
  223. handling_nodes[1] = dst_node;
  224. return 2;
  225. }
  226. else
  227. {
  228. STARPU_ASSERT(max_len >= 1);
  229. src_nodes[0] = src_node;
  230. dst_nodes[0] = dst_node;
  231. handling_nodes[0] = handling_node;
  232. #ifndef HAVE_CUDA_MEMCPY_PEER
  233. STARPU_ASSERT(!(mode & STARPU_R) || starpu_node_get_kind(src_node) != STARPU_CUDA_RAM || starpu_node_get_kind(dst_node) != STARPU_CUDA_RAM);
  234. #endif
  235. return 1;
  236. }
  237. }
  238. /* handle->lock should be taken. r is returned locked. The node parameter
  239. * indicate either the source of the request, or the destination for a
  240. * write-only request. */
  241. static struct _starpu_data_request *_starpu_search_existing_data_request(struct _starpu_data_replicate *replicate, unsigned node, enum starpu_access_mode mode, unsigned is_prefetch)
  242. {
  243. struct _starpu_data_request *r;
  244. r = replicate->request[node];
  245. if (r)
  246. {
  247. _starpu_spin_checklocked(&r->handle->header_lock);
  248. _starpu_spin_lock(&r->lock);
  249. /* perhaps we need to "upgrade" the request */
  250. if (is_prefetch < r->prefetch)
  251. _starpu_update_prefetch_status(r);
  252. if (mode & STARPU_R)
  253. {
  254. /* in case the exisiting request did not imply a memory
  255. * transfer yet, we have to take a second refcnt now
  256. * for the source, in addition to the refcnt for the
  257. * destination
  258. * (so that the source remains valid) */
  259. if (!(r->mode & STARPU_R))
  260. {
  261. replicate->refcnt++;
  262. replicate->handle->busy_count++;
  263. }
  264. r->mode = (enum starpu_access_mode) ((int) r->mode | (int) STARPU_R);
  265. }
  266. if (mode & STARPU_W)
  267. r->mode = (enum starpu_access_mode) ((int) r->mode | (int) STARPU_W);
  268. }
  269. return r;
  270. }
  271. /*
  272. * This function is called when the data is needed on the local node, this
  273. * returns a pointer to the local copy
  274. *
  275. * R STARPU_W STARPU_RW
  276. * Owner OK OK OK
  277. * Shared OK 1 1
  278. * Invalid 2 3 4
  279. *
  280. * case 1 : shared + (read)write :
  281. * no data copy but shared->Invalid/Owner
  282. * case 2 : invalid + read :
  283. * data copy + invalid->shared + owner->shared (STARPU_ASSERT(there is a valid))
  284. * case 3 : invalid + write :
  285. * no data copy + invalid->owner + (owner,shared)->invalid
  286. * case 4 : invalid + R/STARPU_W :
  287. * data copy + if (STARPU_W) (invalid->owner + owner->invalid)
  288. * else (invalid,owner->shared)
  289. */
  290. struct _starpu_data_request *_starpu_create_request_to_fetch_data(starpu_data_handle_t handle,
  291. struct _starpu_data_replicate *dst_replicate,
  292. enum starpu_access_mode mode, unsigned is_prefetch,
  293. unsigned async,
  294. void (*callback_func)(void *), void *callback_arg)
  295. {
  296. /* This function is called with handle's header lock taken */
  297. _starpu_spin_checklocked(&handle->header_lock);
  298. unsigned requesting_node = dst_replicate->memory_node;
  299. if (dst_replicate->state != STARPU_INVALID)
  300. {
  301. #ifdef STARPU_MEMORY_STATS
  302. enum _starpu_cache_state old_state = dst_replicate->state;
  303. #endif
  304. /* the data is already available so we can stop */
  305. _starpu_update_data_state(handle, dst_replicate, mode);
  306. _starpu_msi_cache_hit(requesting_node);
  307. #ifdef STARPU_MEMORY_STATS
  308. _starpu_memory_handle_stats_cache_hit(handle, requesting_node);
  309. /* XXX Broken ? */
  310. if (old_state == STARPU_SHARED
  311. && dst_replicate->state == STARPU_OWNER)
  312. _starpu_memory_handle_stats_shared_to_owner(handle, requesting_node);
  313. #endif
  314. _starpu_memchunk_recently_used(dst_replicate->mc, requesting_node);
  315. _starpu_spin_unlock(&handle->header_lock);
  316. if (callback_func)
  317. callback_func(callback_arg);
  318. _STARPU_LOG_OUT_TAG("data available");
  319. return NULL;
  320. }
  321. _starpu_msi_cache_miss(requesting_node);
  322. /* the only remaining situation is that the local copy was invalid */
  323. STARPU_ASSERT(dst_replicate->state == STARPU_INVALID);
  324. /* find someone who already has the data */
  325. unsigned src_node = 0;
  326. /* if the data is in write only mode, there is no need for a source */
  327. if (mode & STARPU_R)
  328. {
  329. src_node = _starpu_select_src_node(handle, requesting_node);
  330. STARPU_ASSERT(src_node != requesting_node);
  331. }
  332. /* We can safely assume that there won't be more than 2 hops in the
  333. * current implementation */
  334. unsigned src_nodes[4], dst_nodes[4], handling_nodes[4];
  335. int nhops = determine_request_path(handle, src_node, requesting_node, mode, 4,
  336. src_nodes, dst_nodes, handling_nodes);
  337. STARPU_ASSERT(nhops >= 1 && nhops <= 4);
  338. struct _starpu_data_request *requests[nhops];
  339. /* Did we reuse a request for that hop ? */
  340. int reused_requests[nhops];
  341. /* Construct an array with a list of requests, possibly reusing existing requests */
  342. int hop;
  343. for (hop = 0; hop < nhops; hop++)
  344. {
  345. struct _starpu_data_request *r;
  346. unsigned hop_src_node = src_nodes[hop];
  347. unsigned hop_dst_node = dst_nodes[hop];
  348. unsigned hop_handling_node = handling_nodes[hop];
  349. struct _starpu_data_replicate *hop_src_replicate;
  350. struct _starpu_data_replicate *hop_dst_replicate;
  351. /* Only the first request is independant */
  352. unsigned ndeps = (hop == 0)?0:1;
  353. hop_src_replicate = &handle->per_node[hop_src_node];
  354. hop_dst_replicate = (hop != nhops - 1)?&handle->per_node[hop_dst_node]:dst_replicate;
  355. /* Try to reuse a request if possible */
  356. r = _starpu_search_existing_data_request(hop_dst_replicate,
  357. (mode & STARPU_R)?hop_src_node:hop_dst_node,
  358. mode, is_prefetch);
  359. reused_requests[hop] = !!r;
  360. if (!r)
  361. {
  362. /* Create a new request if there was no request to reuse */
  363. r = _starpu_create_data_request(handle, hop_src_replicate,
  364. hop_dst_replicate, hop_handling_node,
  365. mode, ndeps, is_prefetch);
  366. }
  367. requests[hop] = r;
  368. }
  369. /* Chain these requests */
  370. for (hop = 0; hop < nhops; hop++)
  371. {
  372. struct _starpu_data_request *r;
  373. r = requests[hop];
  374. if (hop != nhops - 1)
  375. {
  376. if (!reused_requests[hop + 1])
  377. {
  378. r->next_req[r->next_req_count++] = requests[hop + 1];
  379. STARPU_ASSERT(r->next_req_count <= STARPU_MAXNODES);
  380. }
  381. }
  382. else
  383. /* The last request will perform the callback after termination */
  384. _starpu_data_request_append_callback(r, callback_func, callback_arg);
  385. if (reused_requests[hop])
  386. _starpu_spin_unlock(&r->lock);
  387. }
  388. if (!async)
  389. requests[nhops - 1]->refcnt++;
  390. /* we only submit the first request, the remaining will be
  391. * automatically submitted afterward */
  392. if (!reused_requests[0])
  393. _starpu_post_data_request(requests[0], handling_nodes[0]);
  394. return requests[nhops - 1];
  395. }
  396. int _starpu_fetch_data_on_node(starpu_data_handle_t handle, struct _starpu_data_replicate *dst_replicate,
  397. enum starpu_access_mode mode, unsigned detached, unsigned async,
  398. void (*callback_func)(void *), void *callback_arg)
  399. {
  400. unsigned local_node = _starpu_memory_node_get_local_key();
  401. _STARPU_LOG_IN();
  402. while (_starpu_spin_trylock(&handle->header_lock))
  403. _starpu_datawizard_progress(local_node, 1);
  404. if (!detached)
  405. {
  406. /* Take a reference which will be released by _starpu_release_data_on_node */
  407. dst_replicate->refcnt++;
  408. dst_replicate->handle->busy_count++;
  409. }
  410. struct _starpu_data_request *r;
  411. r = _starpu_create_request_to_fetch_data(handle, dst_replicate, mode,
  412. detached, async, callback_func, callback_arg);
  413. /* If no request was created, the handle was already up-to-date on the
  414. * node. In this case, _starpu_create_request_to_fetch_data has already
  415. * unlocked the header. */
  416. if (!r)
  417. return 0;
  418. _starpu_spin_unlock(&handle->header_lock);
  419. int ret = async?0:_starpu_wait_data_request_completion(r, 1);
  420. _STARPU_LOG_OUT();
  421. return ret;
  422. }
  423. static int prefetch_data_on_node(starpu_data_handle_t handle, struct _starpu_data_replicate *replicate, enum starpu_access_mode mode)
  424. {
  425. return _starpu_fetch_data_on_node(handle, replicate, mode, 1, 1, NULL, NULL);
  426. }
  427. static int fetch_data(starpu_data_handle_t handle, struct _starpu_data_replicate *replicate, enum starpu_access_mode mode)
  428. {
  429. return _starpu_fetch_data_on_node(handle, replicate, mode, 0, 0, NULL, NULL);
  430. }
  431. uint32_t _starpu_get_data_refcnt(starpu_data_handle_t handle, unsigned node)
  432. {
  433. return handle->per_node[node].refcnt;
  434. }
  435. size_t _starpu_data_get_size(starpu_data_handle_t handle)
  436. {
  437. return handle->ops->get_size(handle);
  438. }
  439. uint32_t _starpu_data_get_footprint(starpu_data_handle_t handle)
  440. {
  441. return handle->footprint;
  442. }
  443. /* in case the data was accessed on a write mode, do not forget to
  444. * make it accessible again once it is possible ! */
  445. void _starpu_release_data_on_node(starpu_data_handle_t handle, uint32_t default_wt_mask, struct _starpu_data_replicate *replicate)
  446. {
  447. uint32_t wt_mask;
  448. wt_mask = default_wt_mask | handle->wt_mask;
  449. wt_mask &= (1<<starpu_memory_nodes_get_count())-1;
  450. /* Note that it is possible that there is no valid copy of the data (if
  451. * starpu_data_invalidate was called for instance). In that case, we do
  452. * not enforce any write-through mechanism. */
  453. unsigned memory_node = replicate->memory_node;
  454. if (replicate->state != STARPU_INVALID && handle->current_mode & STARPU_W)
  455. if ((wt_mask & ~(1<<memory_node)))
  456. _starpu_write_through_data(handle, memory_node, wt_mask);
  457. unsigned local_node = _starpu_memory_node_get_local_key();
  458. while (_starpu_spin_trylock(&handle->header_lock))
  459. _starpu_datawizard_progress(local_node, 1);
  460. /* Release refcnt taken by fetch_data_on_node */
  461. replicate->refcnt--;
  462. STARPU_ASSERT_MSG(replicate->refcnt >= 0, "handle %p released too many times", handle);
  463. STARPU_ASSERT_MSG(handle->busy_count > 0, "handle %p released too many times", handle);
  464. handle->busy_count--;
  465. if (!_starpu_notify_data_dependencies(handle))
  466. _starpu_spin_unlock(&handle->header_lock);
  467. }
  468. static void _starpu_set_data_requested_flag_if_needed(struct _starpu_data_replicate *replicate)
  469. {
  470. // XXX : this is just a hint, so we don't take the lock ...
  471. // _starpu_spin_lock(&handle->header_lock);
  472. if (replicate->state == STARPU_INVALID)
  473. {
  474. unsigned dst_node = replicate->memory_node;
  475. replicate->requested[dst_node] = 1;
  476. }
  477. // _starpu_spin_unlock(&handle->header_lock);
  478. }
  479. int starpu_prefetch_task_input_on_node(struct starpu_task *task, unsigned node)
  480. {
  481. unsigned nbuffers = task->cl->nbuffers;
  482. unsigned index;
  483. for (index = 0; index < nbuffers; index++)
  484. {
  485. starpu_data_handle_t handle = STARPU_TASK_GET_HANDLE(task, index);
  486. enum starpu_access_mode mode = STARPU_CODELET_GET_MODE(task->cl, index);
  487. if (mode & (STARPU_SCRATCH|STARPU_REDUX))
  488. continue;
  489. struct _starpu_data_replicate *replicate = &handle->per_node[node];
  490. prefetch_data_on_node(handle, replicate, mode);
  491. _starpu_set_data_requested_flag_if_needed(replicate);
  492. }
  493. return 0;
  494. }
  495. static struct _starpu_data_replicate *get_replicate(starpu_data_handle_t handle, enum starpu_access_mode mode, int workerid, unsigned local_memory_node)
  496. {
  497. if (mode & (STARPU_SCRATCH|STARPU_REDUX))
  498. return &handle->per_worker[workerid];
  499. else
  500. /* That's a "normal" buffer (R/W) */
  501. return &handle->per_node[local_memory_node];
  502. }
  503. int _starpu_fetch_task_input(struct _starpu_job *j, uint32_t mask)
  504. {
  505. _STARPU_TRACE_START_FETCH_INPUT(NULL);
  506. int profiling = starpu_profiling_status_get();
  507. struct starpu_task *task = j->task;
  508. if (profiling && task->profiling_info)
  509. _starpu_clock_gettime(&task->profiling_info->acquire_data_start_time);
  510. struct starpu_buffer_descr *descrs = _STARPU_JOB_GET_ORDERED_BUFFERS(j);
  511. unsigned nbuffers = task->cl->nbuffers;
  512. unsigned local_memory_node = _starpu_memory_node_get_local_key();
  513. int workerid = starpu_worker_get_id();
  514. unsigned index;
  515. for (index = 0; index < nbuffers; index++)
  516. {
  517. int ret;
  518. starpu_data_handle_t handle = descrs[index].handle;
  519. enum starpu_access_mode mode = descrs[index].mode;
  520. struct _starpu_data_replicate *local_replicate;
  521. if (index && descrs[index-1].handle == descrs[index].handle)
  522. /* We have already took this data, skip it. This
  523. * depends on ordering putting writes before reads, see
  524. * _starpu_compar_handles */
  525. continue;
  526. local_replicate = get_replicate(handle, mode, workerid, local_memory_node);
  527. ret = fetch_data(handle, local_replicate, mode);
  528. if (STARPU_UNLIKELY(ret))
  529. goto enomem;
  530. }
  531. /* Now that we have taken the data locks in locking order, fill the codelet interfaces in function order. */
  532. for (index = 0; index < nbuffers; index++)
  533. {
  534. starpu_data_handle_t handle = STARPU_TASK_GET_HANDLE(task, index);
  535. enum starpu_access_mode mode = STARPU_CODELET_GET_MODE(task->cl, index);
  536. struct _starpu_data_replicate *local_replicate;
  537. local_replicate = get_replicate(handle, mode, workerid, local_memory_node);
  538. _STARPU_TASK_SET_INTERFACE(task , local_replicate->data_interface, index);
  539. if (mode & STARPU_REDUX)
  540. {
  541. /* If the replicate was not initialized yet, we have to do it now */
  542. if (!local_replicate->initialized)
  543. _starpu_redux_init_data_replicate(handle, local_replicate, workerid);
  544. }
  545. }
  546. if (profiling && task->profiling_info)
  547. _starpu_clock_gettime(&task->profiling_info->acquire_data_end_time);
  548. _STARPU_TRACE_END_FETCH_INPUT(NULL);
  549. return 0;
  550. enomem:
  551. _STARPU_TRACE_END_FETCH_INPUT(NULL);
  552. /* try to unreference all the input that were successfully taken */
  553. /* XXX broken ... */
  554. _STARPU_DISP("something went wrong with buffer %u\n", index);
  555. //push_codelet_output(task, index, mask);
  556. _starpu_push_task_output(j, mask);
  557. return -1;
  558. }
  559. void _starpu_push_task_output(struct _starpu_job *j, uint32_t mask)
  560. {
  561. _STARPU_TRACE_START_PUSH_OUTPUT(NULL);
  562. int profiling = starpu_profiling_status_get();
  563. struct starpu_task *task = j->task;
  564. if (profiling && task->profiling_info)
  565. _starpu_clock_gettime(&task->profiling_info->release_data_start_time);
  566. struct starpu_buffer_descr *descrs = _STARPU_JOB_GET_ORDERED_BUFFERS(j);
  567. unsigned nbuffers = task->cl->nbuffers;
  568. int workerid = starpu_worker_get_id();
  569. unsigned local_memory_node = _starpu_memory_node_get_local_key();
  570. unsigned index;
  571. for (index = 0; index < nbuffers; index++)
  572. {
  573. starpu_data_handle_t handle = descrs[index].handle;
  574. enum starpu_access_mode mode = descrs[index].mode;
  575. struct _starpu_data_replicate *local_replicate;
  576. if (index && descrs[index-1].handle == descrs[index].handle)
  577. /* We have already released this data, skip it. This
  578. * depends on ordering putting writes before reads, see
  579. * _starpu_compar_handles */
  580. continue;
  581. local_replicate = get_replicate(handle, mode, workerid, local_memory_node);
  582. /* Keep a reference for future
  583. * _starpu_release_task_enforce_sequential_consistency call */
  584. _starpu_spin_lock(&handle->header_lock);
  585. handle->busy_count++;
  586. _starpu_spin_unlock(&handle->header_lock);
  587. _starpu_release_data_on_node(handle, mask, local_replicate);
  588. }
  589. if (profiling && task->profiling_info)
  590. _starpu_clock_gettime(&task->profiling_info->release_data_end_time);
  591. _STARPU_TRACE_END_PUSH_OUTPUT(NULL);
  592. }
  593. /* NB : this value can only be an indication of the status of a data
  594. at some point, but there is no strong garantee ! */
  595. unsigned _starpu_is_data_present_or_requested(starpu_data_handle_t handle, unsigned node)
  596. {
  597. unsigned ret = 0;
  598. // XXX : this is just a hint, so we don't take the lock ...
  599. // _STARPU_PTHREAD_SPIN_LOCK(&handle->header_lock);
  600. if (handle->per_node[node].state != STARPU_INVALID)
  601. {
  602. ret = 1;
  603. }
  604. else
  605. {
  606. unsigned i;
  607. unsigned nnodes = starpu_memory_nodes_get_count();
  608. for (i = 0; i < nnodes; i++)
  609. {
  610. if (handle->per_node[node].requested[i] || handle->per_node[node].request[i])
  611. ret = 1;
  612. }
  613. }
  614. // _STARPU_PTHREAD_SPIN_UNLOCK(&handle->header_lock);
  615. return ret;
  616. }