data_interface.c 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2009-2014 Université de Bordeaux 1
  4. * Copyright (C) 2010, 2011, 2012, 2013, 2014 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 <stdint.h>
  18. #include <datawizard/datawizard.h>
  19. #include <core/dependencies/data_concurrency.h>
  20. #include <common/uthash.h>
  21. #include <common/starpu_spinlock.h>
  22. #include <core/task.h>
  23. #include <core/workers.h>
  24. #include <datawizard/memstats.h>
  25. /* Entry in the `registered_handles' hash table. */
  26. struct handle_entry
  27. {
  28. UT_hash_handle hh;
  29. void *pointer;
  30. starpu_data_handle_t handle;
  31. };
  32. /* Hash table mapping host pointers to data handles. */
  33. static struct handle_entry *registered_handles;
  34. static struct _starpu_spinlock registered_handles_lock;
  35. static int _data_interface_number = STARPU_MAX_INTERFACE_ID;
  36. /* Entry in the `registered_tag_handles' hash table. */
  37. struct handle_tag_entry
  38. {
  39. UT_hash_handle hh;
  40. int tag;
  41. starpu_data_handle_t handle;
  42. };
  43. /* Hash table mapping host tags to data handles. */
  44. static struct handle_tag_entry *registered_tag_handles;
  45. static struct _starpu_spinlock registered_tag_handles_lock;
  46. static void _starpu_data_unregister(starpu_data_handle_t handle, unsigned coherent, unsigned nowait);
  47. void _starpu_data_interface_init(void)
  48. {
  49. _starpu_spin_init(&registered_handles_lock);
  50. _starpu_spin_init(&registered_tag_handles_lock);
  51. }
  52. void _starpu_data_interface_shutdown()
  53. {
  54. struct handle_entry *entry, *tmp;
  55. _starpu_spin_destroy(&registered_handles_lock);
  56. HASH_ITER(hh, registered_handles, entry, tmp)
  57. {
  58. HASH_DEL(registered_handles, entry);
  59. free(entry);
  60. }
  61. registered_handles = NULL;
  62. struct handle_tag_entry *tag_entry, *tag_tmp;
  63. _starpu_spin_destroy(&registered_tag_handles_lock);
  64. HASH_ITER(hh, registered_tag_handles, tag_entry, tag_tmp)
  65. {
  66. HASH_DEL(registered_tag_handles, tag_entry);
  67. free(tag_entry);
  68. }
  69. registered_tag_handles = NULL;
  70. }
  71. struct starpu_data_interface_ops *_starpu_data_interface_get_ops(unsigned interface_id)
  72. {
  73. switch (interface_id)
  74. {
  75. case STARPU_MATRIX_INTERFACE_ID:
  76. return &starpu_interface_matrix_ops;
  77. case STARPU_BLOCK_INTERFACE_ID:
  78. return &starpu_interface_block_ops;
  79. case STARPU_VECTOR_INTERFACE_ID:
  80. return &starpu_interface_vector_ops;
  81. case STARPU_CSR_INTERFACE_ID:
  82. return &starpu_interface_csr_ops;
  83. case STARPU_BCSR_INTERFACE_ID:
  84. return &starpu_interface_bcsr_ops;
  85. case STARPU_VARIABLE_INTERFACE_ID:
  86. return &starpu_interface_variable_ops;
  87. case STARPU_VOID_INTERFACE_ID:
  88. return &starpu_interface_void_ops;
  89. case STARPU_MULTIFORMAT_INTERFACE_ID:
  90. return &starpu_interface_multiformat_ops;
  91. default:
  92. STARPU_ABORT();
  93. return NULL;
  94. }
  95. }
  96. /* Register the mapping from PTR to HANDLE. If PTR is already mapped to
  97. * some handle, the new mapping shadows the previous one. */
  98. void _starpu_data_register_ram_pointer(starpu_data_handle_t handle, void *ptr)
  99. {
  100. struct handle_entry *entry;
  101. entry = (struct handle_entry *) malloc(sizeof(*entry));
  102. STARPU_ASSERT(entry != NULL);
  103. entry->pointer = ptr;
  104. entry->handle = handle;
  105. _starpu_spin_lock(&registered_handles_lock);
  106. HASH_ADD_PTR(registered_handles, pointer, entry);
  107. _starpu_spin_unlock(&registered_handles_lock);
  108. }
  109. starpu_data_handle_t starpu_data_lookup(const void *ptr)
  110. {
  111. starpu_data_handle_t result;
  112. _starpu_spin_lock(&registered_handles_lock);
  113. {
  114. struct handle_entry *entry;
  115. HASH_FIND_PTR(registered_handles, &ptr, entry);
  116. if(STARPU_UNLIKELY(entry == NULL))
  117. result = NULL;
  118. else
  119. result = entry->handle;
  120. }
  121. _starpu_spin_unlock(&registered_handles_lock);
  122. return result;
  123. }
  124. int
  125. _starpu_data_is_multiformat_handle(starpu_data_handle_t handle)
  126. {
  127. return handle->ops->is_multiformat;
  128. }
  129. /*
  130. * Start monitoring a piece of data
  131. */
  132. static void _starpu_register_new_data(starpu_data_handle_t handle,
  133. unsigned home_node, uint32_t wt_mask)
  134. {
  135. void *ptr;
  136. STARPU_ASSERT(handle);
  137. /* initialize the new lock */
  138. handle->req_list = _starpu_data_requester_list_new();
  139. handle->refcnt = 0;
  140. handle->busy_count = 0;
  141. handle->busy_waiting = 0;
  142. STARPU_PTHREAD_MUTEX_INIT(&handle->busy_mutex, NULL);
  143. STARPU_PTHREAD_COND_INIT(&handle->busy_cond, NULL);
  144. _starpu_spin_init(&handle->header_lock);
  145. /* first take care to properly lock the data */
  146. _starpu_spin_lock(&handle->header_lock);
  147. /* there is no hierarchy yet */
  148. handle->nchildren = 0;
  149. handle->root_handle = handle;
  150. handle->father_handle = NULL;
  151. handle->sibling_index = 0; /* could be anything for the root */
  152. handle->depth = 1; /* the tree is just a node yet */
  153. handle->rank = -1; /* invalid until set */
  154. handle->tag = -1; /* invalid until set */
  155. handle->is_not_important = 0;
  156. handle->sequential_consistency =
  157. starpu_data_get_default_sequential_consistency_flag();
  158. STARPU_PTHREAD_MUTEX_INIT(&handle->sequential_consistency_mutex, NULL);
  159. handle->last_submitted_mode = STARPU_R;
  160. handle->last_sync_task = NULL;
  161. handle->last_submitted_accessors = NULL;
  162. handle->post_sync_tasks = NULL;
  163. /* Tell helgrind that the race in _starpu_unlock_post_sync_tasks is fine */
  164. STARPU_HG_DISABLE_CHECKING(handle->post_sync_tasks_cnt);
  165. handle->post_sync_tasks_cnt = 0;
  166. /* By default, there are no methods available to perform a reduction */
  167. handle->redux_cl = NULL;
  168. handle->init_cl = NULL;
  169. handle->reduction_refcnt = 0;
  170. handle->reduction_req_list = _starpu_data_requester_list_new();
  171. #ifdef STARPU_USE_FXT
  172. handle->last_submitted_ghost_sync_id_is_valid = 0;
  173. handle->last_submitted_ghost_sync_id = 0;
  174. handle->last_submitted_ghost_accessors_id = NULL;
  175. #endif
  176. handle->wt_mask = wt_mask;
  177. /* Store some values directly in the handle not to recompute them all
  178. * the time. */
  179. handle->footprint = _starpu_compute_data_footprint(handle);
  180. handle->home_node = home_node;
  181. /* that new data is invalid from all nodes perpective except for the
  182. * home node */
  183. unsigned node;
  184. for (node = 0; node < STARPU_MAXNODES; node++)
  185. {
  186. struct _starpu_data_replicate *replicate;
  187. replicate = &handle->per_node[node];
  188. replicate->memory_node = node;
  189. replicate->relaxed_coherency = 0;
  190. replicate->refcnt = 0;
  191. if (node == home_node)
  192. {
  193. /* this is the home node with the only valid copy */
  194. replicate->state = STARPU_OWNER;
  195. replicate->allocated = 1;
  196. replicate->automatically_allocated = 0;
  197. }
  198. else
  199. {
  200. /* the value is not available here yet */
  201. replicate->state = STARPU_INVALID;
  202. replicate->allocated = 0;
  203. }
  204. }
  205. unsigned worker;
  206. unsigned nworkers = starpu_worker_get_count();
  207. for (worker = 0; worker < nworkers; worker++)
  208. {
  209. struct _starpu_data_replicate *replicate;
  210. replicate = &handle->per_worker[worker];
  211. replicate->allocated = 0;
  212. replicate->automatically_allocated = 0;
  213. replicate->state = STARPU_INVALID;
  214. replicate->refcnt = 0;
  215. replicate->handle = handle;
  216. for (node = 0; node < STARPU_MAXNODES; node++)
  217. {
  218. replicate->requested[node] = 0;
  219. replicate->request[node] = NULL;
  220. }
  221. /* Assuming being used for SCRATCH for now, patched when entering REDUX mode */
  222. replicate->relaxed_coherency = 1;
  223. replicate->initialized = 0;
  224. replicate->memory_node = starpu_worker_get_memory_node(worker);
  225. /* duplicate the content of the interface on node 0 */
  226. memcpy(replicate->data_interface, handle->per_node[0].data_interface, handle->ops->interface_size);
  227. }
  228. /* now the data is available ! */
  229. _starpu_spin_unlock(&handle->header_lock);
  230. ptr = starpu_data_handle_to_pointer(handle, STARPU_MAIN_RAM);
  231. if (ptr != NULL)
  232. {
  233. _starpu_data_register_ram_pointer(handle, ptr);
  234. }
  235. }
  236. int _starpu_data_handle_init(starpu_data_handle_t handle, struct starpu_data_interface_ops *interface_ops, unsigned int mf_node)
  237. {
  238. unsigned node;
  239. unsigned worker;
  240. /* Tell helgrind that our access to busy_count in
  241. * starpu_data_unregister is actually safe */
  242. STARPU_HG_DISABLE_CHECKING(handle->busy_count);
  243. handle->ops = interface_ops;
  244. handle->mf_node = mf_node;
  245. size_t interfacesize = interface_ops->interface_size;
  246. _starpu_memory_stats_init(handle);
  247. for (node = 0; node < STARPU_MAXNODES; node++)
  248. {
  249. _starpu_memory_stats_init_per_node(handle, node);
  250. struct _starpu_data_replicate *replicate;
  251. replicate = &handle->per_node[node];
  252. /* relaxed_coherency = 0 */
  253. replicate->handle = handle;
  254. replicate->data_interface = calloc(1, interfacesize);
  255. STARPU_ASSERT(replicate->data_interface);
  256. }
  257. unsigned nworkers = starpu_worker_get_count();
  258. for (worker = 0; worker < nworkers; worker++)
  259. {
  260. struct _starpu_data_replicate *replicate;
  261. replicate = &handle->per_worker[worker];
  262. replicate->handle = handle;
  263. replicate->data_interface = calloc(1, interfacesize);
  264. STARPU_ASSERT(replicate->data_interface);
  265. }
  266. handle->tag = -1;
  267. handle->rank = -1;
  268. return 0;
  269. }
  270. static
  271. starpu_data_handle_t _starpu_data_handle_allocate(struct starpu_data_interface_ops *interface_ops, unsigned int mf_node)
  272. {
  273. starpu_data_handle_t handle = (starpu_data_handle_t) calloc(1, sizeof(struct _starpu_data_state));
  274. STARPU_ASSERT(handle);
  275. _starpu_data_handle_init(handle, interface_ops, mf_node);
  276. return handle;
  277. }
  278. void starpu_data_register(starpu_data_handle_t *handleptr, unsigned home_node,
  279. void *data_interface,
  280. struct starpu_data_interface_ops *ops)
  281. {
  282. starpu_data_handle_t handle = _starpu_data_handle_allocate(ops, home_node);
  283. STARPU_ASSERT(handleptr);
  284. *handleptr = handle;
  285. /* fill the interface fields with the appropriate method */
  286. STARPU_ASSERT(ops->register_data_handle);
  287. ops->register_data_handle(handle, home_node, data_interface);
  288. _starpu_register_new_data(handle, home_node, 0);
  289. }
  290. void starpu_data_register_same(starpu_data_handle_t *handledst, starpu_data_handle_t handlesrc)
  291. {
  292. void *local_interface = starpu_data_get_interface_on_node(handlesrc, STARPU_MAIN_RAM);
  293. starpu_data_register(handledst, -1, local_interface, handlesrc->ops);
  294. }
  295. void *starpu_data_handle_to_pointer(starpu_data_handle_t handle, unsigned node)
  296. {
  297. /* Check whether the operation is supported and the node has actually
  298. * been allocated. */
  299. if (handle->ops->handle_to_pointer
  300. && starpu_data_test_if_allocated_on_node(handle, node))
  301. {
  302. return handle->ops->handle_to_pointer(handle, node);
  303. }
  304. return NULL;
  305. }
  306. void *starpu_data_get_local_ptr(starpu_data_handle_t handle)
  307. {
  308. return starpu_data_handle_to_pointer(handle,
  309. _starpu_memory_node_get_local_key());
  310. }
  311. int starpu_data_get_rank(starpu_data_handle_t handle)
  312. {
  313. return handle->rank;
  314. }
  315. int _starpu_data_set_rank(starpu_data_handle_t handle, int rank)
  316. {
  317. handle->rank = rank;
  318. return 0;
  319. }
  320. int starpu_data_set_rank(starpu_data_handle_t handle, int rank)
  321. {
  322. _STARPU_DISP("Warning: You should call starpu_mpi_data_register which will insure MPI cache will be cleared when unregistering the data\n");
  323. return _starpu_data_set_rank(handle, rank);
  324. }
  325. int starpu_data_get_tag(starpu_data_handle_t handle)
  326. {
  327. return handle->tag;
  328. }
  329. starpu_data_handle_t _starpu_data_get_data_handle_from_tag(int tag)
  330. {
  331. struct handle_tag_entry *ret;
  332. _starpu_spin_lock(&registered_tag_handles_lock);
  333. HASH_FIND_INT(registered_tag_handles, &tag, ret);
  334. _starpu_spin_unlock(&registered_tag_handles_lock);
  335. if (ret)
  336. {
  337. return ret->handle;
  338. }
  339. else
  340. {
  341. return NULL;
  342. }
  343. }
  344. int _starpu_data_set_tag(starpu_data_handle_t handle, int tag)
  345. {
  346. struct handle_tag_entry *entry;
  347. entry = (struct handle_tag_entry *) malloc(sizeof(*entry));
  348. STARPU_ASSERT(entry != NULL);
  349. STARPU_ASSERT_MSG(!(_starpu_data_get_data_handle_from_tag(tag)),
  350. "There is already a data handle %p registered with the tag %d\n", _starpu_data_get_data_handle_from_tag(tag), tag);
  351. entry->tag = tag;
  352. entry->handle = handle;
  353. _starpu_spin_lock(&registered_tag_handles_lock);
  354. HASH_ADD_INT(registered_tag_handles, tag, entry);
  355. _starpu_spin_unlock(&registered_tag_handles_lock);
  356. handle->tag = tag;
  357. return 0;
  358. }
  359. int starpu_data_set_tag(starpu_data_handle_t handle, int tag)
  360. {
  361. _STARPU_DISP("Warning: You should call starpu_mpi_data_register which will insure MPI cache will be cleared when unregistering the data\n");
  362. return _starpu_data_set_tag(handle, tag);
  363. }
  364. static
  365. int _starpu_data_release_tag(starpu_data_handle_t handle)
  366. {
  367. struct handle_tag_entry *tag_entry;
  368. if (handle->tag != -1)
  369. {
  370. _starpu_spin_lock(&registered_tag_handles_lock);
  371. HASH_FIND_INT(registered_tag_handles, &handle->tag, tag_entry);
  372. STARPU_ASSERT_MSG((tag_entry != NULL),"Data handle %p with tag %d isn't in the hashmap !",handle,handle->tag);
  373. HASH_DEL(registered_tag_handles, tag_entry);
  374. free(tag_entry);
  375. _starpu_spin_unlock(&registered_tag_handles_lock);
  376. }
  377. return 0;
  378. }
  379. struct starpu_data_interface_ops* starpu_data_get_interface_ops(starpu_data_handle_t handle)
  380. {
  381. return handle->ops;
  382. }
  383. /*
  384. * Stop monitoring a piece of data
  385. */
  386. void _starpu_data_unregister_ram_pointer(starpu_data_handle_t handle)
  387. {
  388. const void *ram_ptr = starpu_data_handle_to_pointer(handle, STARPU_MAIN_RAM);
  389. if (ram_ptr != NULL)
  390. {
  391. /* Remove the PTR -> HANDLE mapping. If a mapping from PTR
  392. * to another handle existed before (e.g., when using
  393. * filters), it becomes visible again. */
  394. struct handle_entry *entry;
  395. _starpu_spin_lock(&registered_handles_lock);
  396. HASH_FIND_PTR(registered_handles, &ram_ptr, entry);
  397. STARPU_ASSERT(entry != NULL);
  398. HASH_DEL(registered_handles, entry);
  399. free(entry);
  400. _starpu_spin_unlock(&registered_handles_lock);
  401. }
  402. }
  403. void _starpu_data_free_interfaces(starpu_data_handle_t handle)
  404. {
  405. unsigned node;
  406. unsigned worker;
  407. unsigned nworkers = starpu_worker_get_count();
  408. for (node = 0; node < STARPU_MAXNODES; node++)
  409. free(handle->per_node[node].data_interface);
  410. for (worker = 0; worker < nworkers; worker++)
  411. free(handle->per_worker[worker].data_interface);
  412. }
  413. struct _starpu_unregister_callback_arg
  414. {
  415. unsigned memory_node;
  416. starpu_data_handle_t handle;
  417. unsigned terminated;
  418. starpu_pthread_mutex_t mutex;
  419. starpu_pthread_cond_t cond;
  420. };
  421. /* Check whether we should tell starpu_data_unregister that the data handle is
  422. * not busy any more.
  423. * The header is supposed to be locked.
  424. * This may free the handle, if it was lazily unregistered (1 is returned in
  425. * that case). The handle pointer thus becomes invalid for the caller.
  426. */
  427. int _starpu_data_check_not_busy(starpu_data_handle_t handle)
  428. {
  429. if (!handle->busy_count && handle->busy_waiting)
  430. {
  431. STARPU_PTHREAD_MUTEX_LOCK(&handle->busy_mutex);
  432. STARPU_PTHREAD_COND_BROADCAST(&handle->busy_cond);
  433. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->busy_mutex);
  434. }
  435. /* The handle has been destroyed in between (eg. this was a temporary
  436. * handle created for a reduction.) */
  437. if (handle->lazy_unregister && handle->busy_count == 0)
  438. {
  439. _starpu_spin_unlock(&handle->header_lock);
  440. _starpu_data_unregister(handle, 0, 1);
  441. /* Warning: in case we unregister the handle, we must be sure
  442. * that the caller will not try to unlock the header after
  443. * !*/
  444. return 1;
  445. }
  446. return 0;
  447. }
  448. static void _starpu_data_unregister_fetch_data_callback(void *_arg)
  449. {
  450. int ret;
  451. struct _starpu_unregister_callback_arg *arg = (struct _starpu_unregister_callback_arg *) _arg;
  452. starpu_data_handle_t handle = arg->handle;
  453. STARPU_ASSERT(handle);
  454. struct _starpu_data_replicate *replicate = &handle->per_node[arg->memory_node];
  455. ret = _starpu_fetch_data_on_node(handle, replicate, STARPU_R, 0, 0, NULL, NULL);
  456. STARPU_ASSERT(!ret);
  457. /* unlock the caller */
  458. STARPU_PTHREAD_MUTEX_LOCK(&arg->mutex);
  459. arg->terminated = 1;
  460. STARPU_PTHREAD_COND_SIGNAL(&arg->cond);
  461. STARPU_PTHREAD_MUTEX_UNLOCK(&arg->mutex);
  462. }
  463. /* Unregister the data handle, perhaps we don't need to update the home_node
  464. * (in that case coherent is set to 0)
  465. * nowait is for internal use when we already know for sure that we won't have to wait.
  466. */
  467. static void _starpu_data_unregister(starpu_data_handle_t handle, unsigned coherent, unsigned nowait)
  468. {
  469. STARPU_ASSERT(handle);
  470. STARPU_ASSERT_MSG(handle->nchildren == 0, "data %p needs to be unpartitioned before unregistration", handle);
  471. STARPU_ASSERT(!(nowait && handle->busy_count != 0));
  472. int sequential_consistency = handle->sequential_consistency;
  473. if (sequential_consistency && !nowait)
  474. {
  475. STARPU_ASSERT_MSG(_starpu_worker_may_perform_blocking_calls(), "starpu_data_unregister must not be called from a task or callback, perhaps you can use starpu_data_unregister_submit instead");
  476. /* If sequential consistency is enabled, wait until data is available */
  477. _starpu_data_wait_until_available(handle, STARPU_RW, "starpu_data_unregister");
  478. }
  479. if (coherent && !nowait)
  480. {
  481. STARPU_ASSERT_MSG(_starpu_worker_may_perform_blocking_calls(), "starpu_data_unregister must not be called from a task or callback, perhaps you can use starpu_data_unregister_submit instead");
  482. /* Fetch data in the home of the data to ensure we have a valid copy
  483. * where we registered it */
  484. int home_node = handle->home_node;
  485. if (home_node >= 0)
  486. {
  487. struct _starpu_unregister_callback_arg arg;
  488. arg.handle = handle;
  489. arg.memory_node = (unsigned)home_node;
  490. arg.terminated = 0;
  491. STARPU_PTHREAD_MUTEX_INIT(&arg.mutex, NULL);
  492. STARPU_PTHREAD_COND_INIT(&arg.cond, NULL);
  493. if (!_starpu_attempt_to_submit_data_request_from_apps(handle, STARPU_R,
  494. _starpu_data_unregister_fetch_data_callback, &arg))
  495. {
  496. /* no one has locked this data yet, so we proceed immediately */
  497. struct _starpu_data_replicate *home_replicate = &handle->per_node[home_node];
  498. int ret = _starpu_fetch_data_on_node(handle, home_replicate, STARPU_R, 0, 0, NULL, NULL);
  499. STARPU_ASSERT(!ret);
  500. }
  501. else
  502. {
  503. STARPU_PTHREAD_MUTEX_LOCK(&arg.mutex);
  504. while (!arg.terminated)
  505. STARPU_PTHREAD_COND_WAIT(&arg.cond, &arg.mutex);
  506. STARPU_PTHREAD_MUTEX_UNLOCK(&arg.mutex);
  507. }
  508. STARPU_PTHREAD_MUTEX_DESTROY(&arg.mutex);
  509. STARPU_PTHREAD_COND_DESTROY(&arg.cond);
  510. _starpu_release_data_on_node(handle, 0, &handle->per_node[home_node]);
  511. }
  512. /* If this handle uses a multiformat interface, we may have to convert
  513. * this piece of data back into the CPU format.
  514. * XXX : This is quite hacky, could we submit a task instead ?
  515. */
  516. if (_starpu_data_is_multiformat_handle(handle) &&
  517. ( starpu_node_get_kind(handle->mf_node) != STARPU_CPU_RAM
  518. && starpu_node_get_kind(handle->mf_node) != STARPU_SCC_RAM
  519. && starpu_node_get_kind(handle->mf_node) != STARPU_SCC_SHM
  520. ))
  521. {
  522. _STARPU_DEBUG("Conversion needed\n");
  523. void *buffers[1];
  524. struct starpu_multiformat_interface *format_interface;
  525. format_interface = (struct starpu_multiformat_interface *) starpu_data_get_interface_on_node(handle, STARPU_MAIN_RAM);
  526. struct starpu_codelet *cl = NULL;
  527. enum starpu_node_kind node_kind = starpu_node_get_kind(handle->mf_node);
  528. switch (node_kind)
  529. {
  530. #ifdef STARPU_USE_CUDA
  531. case STARPU_CUDA_RAM:
  532. {
  533. struct starpu_multiformat_data_interface_ops *mf_ops;
  534. mf_ops = (struct starpu_multiformat_data_interface_ops *) handle->ops->get_mf_ops(format_interface);
  535. cl = mf_ops->cuda_to_cpu_cl;
  536. break;
  537. }
  538. #endif
  539. #ifdef STARPU_USE_OPENCL
  540. case STARPU_OPENCL_RAM:
  541. {
  542. struct starpu_multiformat_data_interface_ops *mf_ops;
  543. mf_ops = (struct starpu_multiformat_data_interface_ops *) handle->ops->get_mf_ops(format_interface);
  544. cl = mf_ops->opencl_to_cpu_cl;
  545. break;
  546. }
  547. #endif
  548. #ifdef STARPU_USE_MIC
  549. case STARPU_MIC_RAM:
  550. {
  551. struct starpu_multiformat_data_interface_ops *mf_ops;
  552. mf_ops = (struct starpu_multiformat_data_interface_ops *) handle->ops->get_mf_ops(format_interface);
  553. cl = mf_ops->mic_to_cpu_cl;
  554. break;
  555. }
  556. #endif
  557. case STARPU_CPU_RAM: /* Impossible ! */
  558. case STARPU_SCC_RAM: /* Impossible ! */
  559. case STARPU_SCC_SHM: /* Impossible ! */
  560. default:
  561. STARPU_ABORT();
  562. }
  563. buffers[0] = format_interface;
  564. _starpu_cl_func_t func = _starpu_task_get_cpu_nth_implementation(cl, 0);
  565. STARPU_ASSERT(func);
  566. func(buffers, NULL);
  567. }
  568. }
  569. _starpu_spin_lock(&handle->header_lock);
  570. if (!coherent)
  571. {
  572. /* Should we postpone the unregister operation ? */
  573. if ((handle->busy_count > 0) && handle->lazy_unregister)
  574. {
  575. _starpu_spin_unlock(&handle->header_lock);
  576. return;
  577. }
  578. }
  579. /* Tell holders of references that we're starting waiting */
  580. handle->busy_waiting = 1;
  581. _starpu_spin_unlock(&handle->header_lock);
  582. /* Wait for all requests to finish (notably WT requests) */
  583. STARPU_PTHREAD_MUTEX_LOCK(&handle->busy_mutex);
  584. while (1) {
  585. /* Here helgrind would shout that this an unprotected access,
  586. * but this is actually fine: all threads who do busy_count--
  587. * are supposed to call _starpu_data_check_not_busy, which will
  588. * wake us up through the busy_mutex/busy_cond. */
  589. if (!handle->busy_count)
  590. break;
  591. /* This is woken by _starpu_data_check_not_busy, always called
  592. * after decrementing busy_count */
  593. STARPU_PTHREAD_COND_WAIT(&handle->busy_cond, &handle->busy_mutex);
  594. }
  595. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->busy_mutex);
  596. /* Wait for finished requests to release the handle */
  597. _starpu_spin_lock(&handle->header_lock);
  598. size_t size = _starpu_data_get_size(handle);
  599. _starpu_data_unregister_ram_pointer(handle);
  600. _starpu_data_free_interfaces(handle);
  601. /* Destroy the data now */
  602. unsigned node;
  603. for (node = 0; node < STARPU_MAXNODES; node++)
  604. {
  605. struct _starpu_data_replicate *local = &handle->per_node[node];
  606. /* free the data copy in a lazy fashion */
  607. if (local->allocated && local->automatically_allocated)
  608. _starpu_request_mem_chunk_removal(handle, local, node, size);
  609. }
  610. unsigned worker;
  611. unsigned nworkers = starpu_worker_get_count();
  612. for (worker = 0; worker < nworkers; worker++)
  613. {
  614. struct _starpu_data_replicate *local = &handle->per_worker[worker];
  615. /* free the data copy in a lazy fashion */
  616. if (local->allocated && local->automatically_allocated)
  617. _starpu_request_mem_chunk_removal(handle, local, starpu_worker_get_memory_node(worker), size);
  618. }
  619. _starpu_memory_stats_free(handle);
  620. _starpu_data_requester_list_delete(handle->req_list);
  621. _starpu_data_requester_list_delete(handle->reduction_req_list);
  622. _starpu_spin_unlock(&handle->header_lock);
  623. _starpu_spin_destroy(&handle->header_lock);
  624. STARPU_PTHREAD_MUTEX_DESTROY(&handle->busy_mutex);
  625. STARPU_PTHREAD_COND_DESTROY(&handle->busy_cond);
  626. STARPU_PTHREAD_MUTEX_DESTROY(&handle->sequential_consistency_mutex);
  627. _starpu_data_release_tag(handle);
  628. free(handle);
  629. }
  630. void starpu_data_unregister(starpu_data_handle_t handle)
  631. {
  632. STARPU_ASSERT_MSG(!handle->lazy_unregister, "data %p can not be unregistered twice", handle);
  633. if (handle->unregister_hook)
  634. {
  635. handle->unregister_hook(handle);
  636. }
  637. _starpu_data_unregister(handle, 1, 0);
  638. }
  639. void starpu_data_unregister_no_coherency(starpu_data_handle_t handle)
  640. {
  641. if (handle->unregister_hook)
  642. {
  643. handle->unregister_hook(handle);
  644. }
  645. _starpu_data_unregister(handle, 0, 0);
  646. }
  647. static void _starpu_data_unregister_submit_cb(void *arg)
  648. {
  649. starpu_data_handle_t handle = arg;
  650. _starpu_spin_lock(&handle->header_lock);
  651. handle->lazy_unregister = 1;
  652. /* The handle should be busy since we are working on it.
  653. * when we releases the handle below, it will be destroyed by
  654. * _starpu_data_check_not_busy */
  655. STARPU_ASSERT(handle->busy_count);
  656. _starpu_spin_unlock(&handle->header_lock);
  657. starpu_data_release_on_node(handle, -1);
  658. }
  659. void starpu_data_unregister_submit(starpu_data_handle_t handle)
  660. {
  661. STARPU_ASSERT_MSG(!handle->lazy_unregister, "data %p can not be unregistered twice", handle);
  662. if (handle->unregister_hook)
  663. {
  664. handle->unregister_hook(handle);
  665. }
  666. /* Wait for all task dependencies on this handle before putting it for free */
  667. starpu_data_acquire_on_node_cb(handle, -1, STARPU_RW, _starpu_data_unregister_submit_cb, handle);
  668. }
  669. static void _starpu_data_invalidate(void *data)
  670. {
  671. starpu_data_handle_t handle = data;
  672. size_t size = _starpu_data_get_size(handle);
  673. _starpu_spin_lock(&handle->header_lock);
  674. unsigned node;
  675. for (node = 0; node < STARPU_MAXNODES; node++)
  676. {
  677. struct _starpu_data_replicate *local = &handle->per_node[node];
  678. if (local->mc && local->allocated && local->automatically_allocated)
  679. {
  680. if (node == STARPU_MAIN_RAM)
  681. _starpu_data_unregister_ram_pointer(handle);
  682. /* free the data copy in a lazy fashion */
  683. _starpu_request_mem_chunk_removal(handle, local, node, size);
  684. }
  685. local->state = STARPU_INVALID;
  686. }
  687. unsigned worker;
  688. unsigned nworkers = starpu_worker_get_count();
  689. for (worker = 0; worker < nworkers; worker++)
  690. {
  691. struct _starpu_data_replicate *local = &handle->per_worker[worker];
  692. if (local->mc && local->allocated && local->automatically_allocated)
  693. /* free the data copy in a lazy fashion */
  694. _starpu_request_mem_chunk_removal(handle, local, starpu_worker_get_memory_node(worker), size);
  695. local->state = STARPU_INVALID;
  696. }
  697. _starpu_spin_unlock(&handle->header_lock);
  698. starpu_data_release_on_node(handle, -1);
  699. }
  700. void starpu_data_invalidate(starpu_data_handle_t handle)
  701. {
  702. STARPU_ASSERT(handle);
  703. starpu_data_acquire_on_node(handle, -1, STARPU_W);
  704. _starpu_data_invalidate(handle);
  705. }
  706. void starpu_data_invalidate_submit(starpu_data_handle_t handle)
  707. {
  708. STARPU_ASSERT(handle);
  709. starpu_data_acquire_on_node_cb(handle, -1, STARPU_W, _starpu_data_invalidate, handle);
  710. }
  711. enum starpu_data_interface_id starpu_data_get_interface_id(starpu_data_handle_t handle)
  712. {
  713. return handle->ops->interfaceid;
  714. }
  715. void *starpu_data_get_interface_on_node(starpu_data_handle_t handle, unsigned memory_node)
  716. {
  717. return handle->per_node[memory_node].data_interface;
  718. }
  719. int starpu_data_interface_get_next_id(void)
  720. {
  721. _data_interface_number += 1;
  722. return _data_interface_number-1;
  723. }
  724. int starpu_data_pack(starpu_data_handle_t handle, void **ptr, starpu_ssize_t *count)
  725. {
  726. STARPU_ASSERT(handle->ops->pack_data);
  727. return handle->ops->pack_data(handle, _starpu_memory_node_get_local_key(), ptr, count);
  728. }
  729. int starpu_data_unpack(starpu_data_handle_t handle, void *ptr, size_t count)
  730. {
  731. STARPU_ASSERT(handle->ops->unpack_data);
  732. int ret;
  733. ret = handle->ops->unpack_data(handle, _starpu_memory_node_get_local_key(), ptr, count);
  734. free(ptr);
  735. return ret;
  736. }
  737. size_t starpu_data_get_size(starpu_data_handle_t handle)
  738. {
  739. return handle->ops->get_size(handle);
  740. }