data_interface.c 27 KB

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