data_interface.c 27 KB

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