data_interface.c 33 KB

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