data_interface.c 30 KB

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