coherency.c 22 KB

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