filters.c 35 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2011 Antoine Lucas
  4. * Copyright (C) 2011-2012,2016-2017 Inria
  5. * Copyright (C) 2008-2017 Université de Bordeaux
  6. * Copyright (C) 2010 Mehdi Juhoor
  7. * Copyright (C) 2010-2013,2015-2018 CNRS
  8. * Copyright (C) 2013 Thibaut Lambert
  9. *
  10. * StarPU is free software; you can redistribute it and/or modify
  11. * it under the terms of the GNU Lesser General Public License as published by
  12. * the Free Software Foundation; either version 2.1 of the License, or (at
  13. * your option) any later version.
  14. *
  15. * StarPU is distributed in the hope that it will be useful, but
  16. * WITHOUT ANY WARRANTY; without even the implied warranty of
  17. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  18. *
  19. * See the GNU Lesser General Public License in COPYING.LGPL for more details.
  20. */
  21. //#define STARPU_VERBOSE
  22. #include <datawizard/filters.h>
  23. #include <datawizard/footprint.h>
  24. #include <datawizard/interfaces/data_interface.h>
  25. #include <core/task.h>
  26. /*
  27. * This function applies a data filter on all the elements of a partition
  28. */
  29. static void map_filter(starpu_data_handle_t root_handle, struct starpu_data_filter *f)
  30. {
  31. /* we need to apply the data filter on all leaf of the tree */
  32. if (root_handle->nchildren == 0)
  33. {
  34. /* this is a leaf */
  35. starpu_data_partition(root_handle, f);
  36. }
  37. else
  38. {
  39. /* try to apply the data filter recursively */
  40. unsigned child;
  41. for (child = 0; child < root_handle->nchildren; child++)
  42. {
  43. starpu_data_handle_t handle_child = starpu_data_get_child(root_handle, child);
  44. map_filter(handle_child, f);
  45. }
  46. }
  47. }
  48. void starpu_data_vmap_filters(starpu_data_handle_t root_handle, unsigned nfilters, va_list pa)
  49. {
  50. unsigned i;
  51. for (i = 0; i < nfilters; i++)
  52. {
  53. struct starpu_data_filter *next_filter;
  54. next_filter = va_arg(pa, struct starpu_data_filter *);
  55. STARPU_ASSERT(next_filter);
  56. map_filter(root_handle, next_filter);
  57. }
  58. }
  59. void starpu_data_map_filters(starpu_data_handle_t root_handle, unsigned nfilters, ...)
  60. {
  61. va_list pa;
  62. va_start(pa, nfilters);
  63. starpu_data_vmap_filters(root_handle, nfilters, pa);
  64. va_end(pa);
  65. }
  66. void fstarpu_data_map_filters(starpu_data_handle_t root_handle, int nfilters, struct starpu_data_filter **filters)
  67. {
  68. int i;
  69. assert(nfilters >= 0);
  70. for (i = 0; i < nfilters; i++)
  71. {
  72. struct starpu_data_filter *next_filter = filters[i];
  73. STARPU_ASSERT(next_filter);
  74. map_filter(root_handle, next_filter);
  75. }
  76. }
  77. int starpu_data_get_nb_children(starpu_data_handle_t handle)
  78. {
  79. return handle->nchildren;
  80. }
  81. starpu_data_handle_t starpu_data_get_child(starpu_data_handle_t handle, unsigned i)
  82. {
  83. STARPU_ASSERT_MSG(handle->nchildren != 0, "Data %p has to be partitioned before accessing children", handle);
  84. STARPU_ASSERT_MSG(i < handle->nchildren, "Invalid child index %u in handle %p, maximum %u", i, handle, handle->nchildren);
  85. return &handle->children[i];
  86. }
  87. /*
  88. * example starpu_data_get_sub_data(starpu_data_handle_t root_handle, 3, 42, 0, 1);
  89. */
  90. starpu_data_handle_t starpu_data_get_sub_data(starpu_data_handle_t root_handle, unsigned depth, ... )
  91. {
  92. va_list pa;
  93. va_start(pa, depth);
  94. starpu_data_handle_t handle = starpu_data_vget_sub_data(root_handle, depth, pa);
  95. va_end(pa);
  96. return handle;
  97. }
  98. starpu_data_handle_t starpu_data_vget_sub_data(starpu_data_handle_t root_handle, unsigned depth, va_list pa )
  99. {
  100. STARPU_ASSERT(root_handle);
  101. starpu_data_handle_t current_handle = root_handle;
  102. /* the variable number of argument must correlate the depth in the tree */
  103. unsigned i;
  104. for (i = 0; i < depth; i++)
  105. {
  106. unsigned next_child;
  107. next_child = va_arg(pa, unsigned);
  108. STARPU_ASSERT_MSG(current_handle->nchildren != 0, "Data %p has to be partitioned before accessing children", current_handle);
  109. STARPU_ASSERT_MSG(next_child < current_handle->nchildren, "Bogus child number %u, data %p only has %u children", next_child, current_handle, current_handle->nchildren);
  110. current_handle = &current_handle->children[next_child];
  111. }
  112. return current_handle;
  113. }
  114. starpu_data_handle_t fstarpu_data_get_sub_data(starpu_data_handle_t root_handle, int depth, int *indices)
  115. {
  116. STARPU_ASSERT(root_handle);
  117. starpu_data_handle_t current_handle = root_handle;
  118. STARPU_ASSERT(depth >= 0);
  119. /* the variable number of argument must correlate the depth in the tree */
  120. int i;
  121. for (i = 0; i < depth; i++)
  122. {
  123. int next_child;
  124. next_child = indices[i];
  125. STARPU_ASSERT(next_child >= 0);
  126. STARPU_ASSERT_MSG(current_handle->nchildren != 0, "Data %p has to be partitioned before accessing children", current_handle);
  127. STARPU_ASSERT_MSG((unsigned) next_child < current_handle->nchildren, "Bogus child number %d, data %p only has %u children", next_child, current_handle, current_handle->nchildren);
  128. current_handle = &current_handle->children[next_child];
  129. }
  130. return current_handle;
  131. }
  132. static unsigned _starpu_data_partition_nparts(starpu_data_handle_t initial_handle, struct starpu_data_filter *f)
  133. {
  134. /* how many parts ? */
  135. if (f->get_nchildren)
  136. return f->get_nchildren(f, initial_handle);
  137. else
  138. return f->nchildren;
  139. }
  140. static void _starpu_data_partition(starpu_data_handle_t initial_handle, starpu_data_handle_t *childrenp, unsigned nparts, struct starpu_data_filter *f, int inherit_state)
  141. {
  142. unsigned i;
  143. unsigned node;
  144. /* first take care to properly lock the data header */
  145. _starpu_spin_lock(&initial_handle->header_lock);
  146. initial_handle->nplans++;
  147. STARPU_ASSERT_MSG(nparts > 0, "Partitioning data %p in 0 piece does not make sense", initial_handle);
  148. /* allocate the children */
  149. if (inherit_state)
  150. {
  151. _STARPU_CALLOC(initial_handle->children, nparts, sizeof(struct _starpu_data_state));
  152. /* this handle now has children */
  153. initial_handle->nchildren = nparts;
  154. }
  155. for (node = 0; node < STARPU_MAXNODES; node++)
  156. {
  157. if (initial_handle->per_node[node].state != STARPU_INVALID)
  158. break;
  159. }
  160. if (node == STARPU_MAXNODES)
  161. {
  162. /* This is lazy allocation, allocate it now in main RAM, so as
  163. * to have somewhere to gather pieces later */
  164. /* FIXME: mark as unevictable! */
  165. int home_node = initial_handle->home_node;
  166. if (home_node < 0 || (starpu_node_get_kind(home_node) != STARPU_CPU_RAM))
  167. home_node = STARPU_MAIN_RAM;
  168. int ret = _starpu_allocate_memory_on_node(initial_handle, &initial_handle->per_node[home_node], 0);
  169. #ifdef STARPU_DEVEL
  170. #warning we should reclaim memory if allocation failed
  171. #endif
  172. STARPU_ASSERT(!ret);
  173. }
  174. for (node = 0; node < STARPU_MAXNODES; node++)
  175. _starpu_data_unregister_ram_pointer(initial_handle, node);
  176. if (nparts && !inherit_state)
  177. {
  178. STARPU_ASSERT_MSG(childrenp, "Passing NULL pointer for parameter childrenp while parameter inherit_state is 0");
  179. }
  180. for (i = 0; i < nparts; i++)
  181. {
  182. starpu_data_handle_t child;
  183. if (inherit_state)
  184. child = &initial_handle->children[i];
  185. else
  186. child = childrenp[i];
  187. STARPU_ASSERT(child);
  188. struct starpu_data_interface_ops *ops;
  189. /* each child may have his own interface type */
  190. /* what's this child's interface ? */
  191. if (f->get_child_ops)
  192. ops = f->get_child_ops(f, i);
  193. else
  194. ops = initial_handle->ops;
  195. _starpu_data_handle_init(child, ops, initial_handle->mf_node);
  196. child->nchildren = 0;
  197. child->nplans = 0;
  198. child->switch_cl = NULL;
  199. child->partitioned = 0;
  200. child->readonly = 0;
  201. child->active = inherit_state;
  202. child->active_ro = 0;
  203. child->mpi_data = initial_handle->mpi_data;
  204. child->root_handle = initial_handle->root_handle;
  205. child->father_handle = initial_handle;
  206. child->active_children = NULL;
  207. child->active_readonly_children = NULL;
  208. child->nactive_readonly_children = 0;
  209. child->nsiblings = nparts;
  210. if (inherit_state)
  211. child->siblings = NULL;
  212. else
  213. child->siblings = childrenp;
  214. child->sibling_index = i;
  215. child->depth = initial_handle->depth + 1;
  216. child->is_not_important = initial_handle->is_not_important;
  217. child->wt_mask = initial_handle->wt_mask;
  218. child->home_node = initial_handle->home_node;
  219. /* initialize the chunk lock */
  220. _starpu_data_requester_list_init(&child->req_list);
  221. _starpu_data_requester_list_init(&child->reduction_req_list);
  222. child->reduction_tmp_handles = NULL;
  223. child->write_invalidation_req = NULL;
  224. child->refcnt = 0;
  225. child->unlocking_reqs = 0;
  226. child->busy_count = 0;
  227. child->busy_waiting = 0;
  228. STARPU_PTHREAD_MUTEX_INIT(&child->busy_mutex, NULL);
  229. STARPU_PTHREAD_COND_INIT(&child->busy_cond, NULL);
  230. child->reduction_refcnt = 0;
  231. _starpu_spin_init(&child->header_lock);
  232. child->sequential_consistency = initial_handle->sequential_consistency;
  233. child->initialized = initial_handle->initialized;
  234. STARPU_PTHREAD_MUTEX_INIT(&child->sequential_consistency_mutex, NULL);
  235. child->last_submitted_mode = STARPU_R;
  236. child->last_sync_task = NULL;
  237. child->last_submitted_accessors.task = NULL;
  238. child->last_submitted_accessors.next = &child->last_submitted_accessors;
  239. child->last_submitted_accessors.prev = &child->last_submitted_accessors;
  240. child->post_sync_tasks = NULL;
  241. /* Tell helgrind that the race in _starpu_unlock_post_sync_tasks is fine */
  242. STARPU_HG_DISABLE_CHECKING(child->post_sync_tasks_cnt);
  243. child->post_sync_tasks_cnt = 0;
  244. /* The methods used for reduction are propagated to the
  245. * children. */
  246. child->redux_cl = initial_handle->redux_cl;
  247. child->init_cl = initial_handle->init_cl;
  248. #ifdef STARPU_USE_FXT
  249. child->last_submitted_ghost_sync_id_is_valid = 0;
  250. child->last_submitted_ghost_sync_id = 0;
  251. child->last_submitted_ghost_accessors_id = NULL;
  252. #endif
  253. if (_starpu_global_arbiter)
  254. /* Just for testing purpose */
  255. starpu_data_assign_arbiter(child, _starpu_global_arbiter);
  256. else
  257. child->arbiter = NULL;
  258. _starpu_data_requester_list_init(&child->arbitered_req_list);
  259. for (node = 0; node < STARPU_MAXNODES; node++)
  260. {
  261. struct _starpu_data_replicate *initial_replicate;
  262. struct _starpu_data_replicate *child_replicate;
  263. initial_replicate = &initial_handle->per_node[node];
  264. child_replicate = &child->per_node[node];
  265. if (inherit_state)
  266. child_replicate->state = initial_replicate->state;
  267. else
  268. child_replicate->state = STARPU_INVALID;
  269. if (inherit_state || !initial_replicate->automatically_allocated)
  270. child_replicate->allocated = initial_replicate->allocated;
  271. else
  272. child_replicate->allocated = 0;
  273. /* Do not allow memory reclaiming within the child for parent bits */
  274. child_replicate->automatically_allocated = 0;
  275. child_replicate->refcnt = 0;
  276. child_replicate->memory_node = node;
  277. child_replicate->relaxed_coherency = 0;
  278. if (inherit_state)
  279. child_replicate->initialized = initial_replicate->initialized;
  280. else
  281. child_replicate->initialized = 0;
  282. /* update the interface */
  283. void *initial_interface = starpu_data_get_interface_on_node(initial_handle, node);
  284. void *child_interface = starpu_data_get_interface_on_node(child, node);
  285. STARPU_ASSERT_MSG(!(!inherit_state && child_replicate->automatically_allocated && child_replicate->allocated), "partition planning is currently not supported when handle has some automatically allocated buffers");
  286. f->filter_func(initial_interface, child_interface, f, i, nparts);
  287. }
  288. child->per_worker = NULL;
  289. child->user_data = NULL;
  290. /* We compute the size and the footprint of the child once and
  291. * store it in the handle */
  292. child->footprint = _starpu_compute_data_footprint(child);
  293. for (node = 0; node < STARPU_MAXNODES; node++)
  294. {
  295. if (starpu_node_get_kind(node) != STARPU_CPU_RAM)
  296. continue;
  297. void *ptr = starpu_data_handle_to_pointer(child, node);
  298. if (ptr != NULL)
  299. _starpu_data_register_ram_pointer(child, ptr);
  300. }
  301. _STARPU_TRACE_HANDLE_DATA_REGISTER(child);
  302. }
  303. /* now let the header */
  304. _starpu_spin_unlock(&initial_handle->header_lock);
  305. }
  306. static
  307. void _starpu_empty_codelet_function(void *buffers[], void *args)
  308. {
  309. (void) buffers; // unused;
  310. (void) args; // unused;
  311. }
  312. void starpu_data_unpartition(starpu_data_handle_t root_handle, unsigned gathering_node)
  313. {
  314. unsigned child;
  315. unsigned worker;
  316. unsigned nworkers = starpu_worker_get_count();
  317. unsigned node;
  318. unsigned sizes[root_handle->nchildren];
  319. void *ptr;
  320. _STARPU_TRACE_START_UNPARTITION(root_handle, gathering_node);
  321. _starpu_spin_lock(&root_handle->header_lock);
  322. STARPU_ASSERT_MSG(root_handle->nchildren != 0, "data %p is not partitioned, can not unpartition it", root_handle);
  323. /* first take all the children lock (in order !) */
  324. for (child = 0; child < root_handle->nchildren; child++)
  325. {
  326. starpu_data_handle_t child_handle = starpu_data_get_child(root_handle, child);
  327. /* make sure the intermediate children is unpartitionned as well */
  328. if (child_handle->nchildren > 0)
  329. starpu_data_unpartition(child_handle, gathering_node);
  330. /* If this is a multiformat handle, we must convert the data now */
  331. #ifdef STARPU_DEVEL
  332. #warning TODO: _starpu_fetch_data_on_node should be doing it
  333. #endif
  334. if (_starpu_data_is_multiformat_handle(child_handle) &&
  335. starpu_node_get_kind(child_handle->mf_node) != STARPU_CPU_RAM)
  336. {
  337. struct starpu_codelet cl =
  338. {
  339. .where = STARPU_CPU,
  340. .cpu_funcs = { _starpu_empty_codelet_function },
  341. .modes = { STARPU_RW },
  342. .nbuffers = 1
  343. };
  344. struct starpu_task *task = starpu_task_create();
  345. task->name = "convert_data";
  346. STARPU_TASK_SET_HANDLE(task, child_handle, 0);
  347. task->cl = &cl;
  348. task->synchronous = 1;
  349. if (_starpu_task_submit_internally(task) != 0)
  350. _STARPU_ERROR("Could not submit the conversion task while unpartitionning\n");
  351. }
  352. int ret;
  353. /* for now we pretend that the RAM is almost unlimited and that gathering
  354. * data should be possible from the node that does the unpartionning ... we
  355. * don't want to have the programming deal with memory shortage at that time,
  356. * really */
  357. /* Acquire the child data on the gathering node. This will trigger collapsing any reduction */
  358. ret = starpu_data_acquire_on_node(child_handle, gathering_node, STARPU_RW);
  359. STARPU_ASSERT(ret == 0);
  360. starpu_data_release_on_node(child_handle, gathering_node);
  361. _starpu_spin_lock(&child_handle->header_lock);
  362. child_handle->busy_waiting = 1;
  363. _starpu_spin_unlock(&child_handle->header_lock);
  364. /* Wait for all requests to finish (notably WT requests) */
  365. STARPU_PTHREAD_MUTEX_LOCK(&child_handle->busy_mutex);
  366. while (1)
  367. {
  368. /* Here helgrind would shout that this an unprotected access,
  369. * but this is actually fine: all threads who do busy_count--
  370. * are supposed to call _starpu_data_check_not_busy, which will
  371. * wake us up through the busy_mutex/busy_cond. */
  372. if (!child_handle->busy_count)
  373. break;
  374. /* This is woken by _starpu_data_check_not_busy, always called
  375. * after decrementing busy_count */
  376. STARPU_PTHREAD_COND_WAIT(&child_handle->busy_cond, &child_handle->busy_mutex);
  377. }
  378. STARPU_PTHREAD_MUTEX_UNLOCK(&child_handle->busy_mutex);
  379. _starpu_spin_lock(&child_handle->header_lock);
  380. sizes[child] = _starpu_data_get_size(child_handle);
  381. if (child_handle->unregister_hook)
  382. {
  383. child_handle->unregister_hook(child_handle);
  384. }
  385. for (node = 0; node < STARPU_MAXNODES; node++)
  386. _starpu_data_unregister_ram_pointer(child_handle, node);
  387. if (child_handle->per_worker)
  388. {
  389. for (worker = 0; worker < nworkers; worker++)
  390. {
  391. struct _starpu_data_replicate *local = &child_handle->per_worker[worker];
  392. STARPU_ASSERT(local->state == STARPU_INVALID);
  393. if (local->allocated && local->automatically_allocated)
  394. _starpu_request_mem_chunk_removal(child_handle, local, starpu_worker_get_memory_node(worker), sizes[child]);
  395. }
  396. }
  397. _starpu_memory_stats_free(child_handle);
  398. }
  399. for (node = 0; node < STARPU_MAXNODES; node++)
  400. {
  401. if (starpu_node_get_kind(node) != STARPU_CPU_RAM)
  402. continue;
  403. ptr = starpu_data_handle_to_pointer(root_handle, node);
  404. if (ptr != NULL)
  405. _starpu_data_register_ram_pointer(root_handle, ptr);
  406. }
  407. /* the gathering_node should now have a valid copy of all the children.
  408. * For all nodes, if the node had all copies and none was locally
  409. * allocated then the data is still valid there, else, it's invalidated
  410. * for the gathering node, if we have some locally allocated data, we
  411. * copy all the children (XXX this should not happen so we just do not
  412. * do anything since this is transparent ?) */
  413. unsigned still_valid[STARPU_MAXNODES];
  414. /* we do 2 passes : the first pass determines wether the data is still
  415. * valid or not, the second pass is needed to choose between STARPU_SHARED and
  416. * STARPU_OWNER */
  417. unsigned nvalids = 0;
  418. /* still valid ? */
  419. for (node = 0; node < STARPU_MAXNODES; node++)
  420. {
  421. struct _starpu_data_replicate *local;
  422. /* until an issue is found the data is assumed to be valid */
  423. unsigned isvalid = 1;
  424. for (child = 0; child < root_handle->nchildren; child++)
  425. {
  426. starpu_data_handle_t child_handle = starpu_data_get_child(root_handle, child);
  427. local = &child_handle->per_node[node];
  428. if (local->state == STARPU_INVALID || local->automatically_allocated == 1)
  429. {
  430. /* One of the bits is missing or is not inside the parent */
  431. isvalid = 0;
  432. }
  433. if (local->mc && local->allocated && local->automatically_allocated)
  434. /* free the child data copy in a lazy fashion */
  435. _starpu_request_mem_chunk_removal(child_handle, local, node, sizes[child]);
  436. }
  437. local = &root_handle->per_node[node];
  438. if (!local->allocated)
  439. /* Even if we have all the bits, if we don't have the
  440. * whole data, it's not valid */
  441. isvalid = 0;
  442. if (!isvalid && local->mc && local->allocated && local->automatically_allocated)
  443. /* free the data copy in a lazy fashion */
  444. _starpu_request_mem_chunk_removal(root_handle, local, node, _starpu_data_get_size(root_handle));
  445. /* if there was no invalid copy, the node still has a valid copy */
  446. still_valid[node] = isvalid;
  447. if (isvalid)
  448. nvalids++;
  449. }
  450. /* either shared or owned */
  451. STARPU_ASSERT(nvalids > 0);
  452. enum _starpu_cache_state newstate = (nvalids == 1)?STARPU_OWNER:STARPU_SHARED;
  453. for (node = 0; node < STARPU_MAXNODES; node++)
  454. {
  455. root_handle->per_node[node].state =
  456. still_valid[node]?newstate:STARPU_INVALID;
  457. }
  458. for (child = 0; child < root_handle->nchildren; child++)
  459. {
  460. starpu_data_handle_t child_handle = starpu_data_get_child(root_handle, child);
  461. _starpu_data_free_interfaces(child_handle);
  462. _starpu_spin_unlock(&child_handle->header_lock);
  463. _starpu_spin_destroy(&child_handle->header_lock);
  464. }
  465. for (child = 0; child < root_handle->nchildren; child++)
  466. {
  467. starpu_data_handle_t child_handle = starpu_data_get_child(root_handle, child);
  468. _starpu_data_clear_implicit(child_handle);
  469. STARPU_PTHREAD_MUTEX_DESTROY(&child_handle->busy_mutex);
  470. STARPU_PTHREAD_COND_DESTROY(&child_handle->busy_cond);
  471. STARPU_PTHREAD_MUTEX_DESTROY(&child_handle->sequential_consistency_mutex);
  472. _STARPU_TRACE_HANDLE_DATA_UNREGISTER(child_handle);
  473. }
  474. /* there is no child anymore */
  475. starpu_data_handle_t children = root_handle->children;
  476. root_handle->children = NULL;
  477. root_handle->nchildren = 0;
  478. root_handle->nplans--;
  479. /* now the parent may be used again so we release the lock */
  480. _starpu_spin_unlock(&root_handle->header_lock);
  481. free(children);
  482. _STARPU_TRACE_END_UNPARTITION(root_handle, gathering_node);
  483. }
  484. void starpu_data_partition(starpu_data_handle_t initial_handle, struct starpu_data_filter *f)
  485. {
  486. unsigned nparts = _starpu_data_partition_nparts(initial_handle, f);
  487. STARPU_ASSERT_MSG(initial_handle->nchildren == 0, "there should not be mutiple filters applied on the same data %p, futher filtering has to be done on children", initial_handle);
  488. STARPU_ASSERT_MSG(initial_handle->nplans == 0, "partition planning and synchronous partitioning is not supported");
  489. initial_handle->children = NULL;
  490. /* Make sure to wait for previous tasks working on the whole data */
  491. starpu_data_acquire_on_node(initial_handle, STARPU_ACQUIRE_NO_NODE, initial_handle->initialized?STARPU_RW:STARPU_W);
  492. starpu_data_release_on_node(initial_handle, STARPU_ACQUIRE_NO_NODE);
  493. _starpu_data_partition(initial_handle, NULL, nparts, f, 1);
  494. }
  495. void starpu_data_partition_plan(starpu_data_handle_t initial_handle, struct starpu_data_filter *f, starpu_data_handle_t *childrenp)
  496. {
  497. unsigned i;
  498. unsigned nparts = _starpu_data_partition_nparts(initial_handle, f);
  499. STARPU_ASSERT_MSG(initial_handle->nchildren == 0, "partition planning and synchronous partitioning is not supported");
  500. STARPU_ASSERT_MSG(initial_handle->sequential_consistency, "partition planning is currently only supported for data with sequential consistency");
  501. struct starpu_codelet *cl = initial_handle->switch_cl;
  502. int home_node = initial_handle->home_node;
  503. starpu_data_handle_t *children;
  504. if (home_node == -1)
  505. /* Nothing better for now */
  506. /* TODO: pass -1, and make _starpu_fetch_nowhere_task_input
  507. * really call _starpu_fetch_data_on_node, and make that update
  508. * the coherency.
  509. */
  510. home_node = STARPU_MAIN_RAM;
  511. _STARPU_MALLOC(children, nparts * sizeof(*children));
  512. for (i = 0; i < nparts; i++)
  513. {
  514. _STARPU_CALLOC(children[i], 1, sizeof(struct _starpu_data_state));
  515. childrenp[i] = children[i];
  516. }
  517. _starpu_data_partition(initial_handle, children, nparts, f, 0);
  518. if (!cl)
  519. {
  520. /* Create a codelet that will make the coherency on the home node */
  521. _STARPU_CALLOC(initial_handle->switch_cl, 1, sizeof(*initial_handle->switch_cl));
  522. cl = initial_handle->switch_cl;
  523. cl->where = STARPU_NOWHERE;
  524. cl->nbuffers = STARPU_VARIABLE_NBUFFERS;
  525. cl->flags = STARPU_CODELET_NOPLANS;
  526. cl->name = "data_partition_switch";
  527. cl->specific_nodes = 1;
  528. }
  529. if (initial_handle->switch_cl_nparts < nparts)
  530. {
  531. /* First initialization, or previous initialization was with fewer parts, enlarge it */
  532. _STARPU_REALLOC(cl->dyn_nodes, (nparts+1) * sizeof(*cl->dyn_nodes));
  533. for (i = initial_handle->switch_cl_nparts; i < nparts+1; i++)
  534. cl->dyn_nodes[i] = home_node;
  535. initial_handle->switch_cl_nparts = nparts;
  536. }
  537. }
  538. void starpu_data_partition_clean(starpu_data_handle_t root_handle, unsigned nparts, starpu_data_handle_t *children)
  539. {
  540. unsigned i;
  541. if (children[0]->active)
  542. {
  543. #ifdef STARPU_DEVEL
  544. #warning FIXME: better choose gathering node
  545. #endif
  546. starpu_data_unpartition_submit(root_handle, nparts, children, STARPU_MAIN_RAM);
  547. }
  548. free(children[0]->siblings);
  549. for (i = 0; i < nparts; i++)
  550. starpu_data_unregister_submit(children[i]);
  551. _starpu_spin_lock(&root_handle->header_lock);
  552. root_handle->nplans--;
  553. _starpu_spin_unlock(&root_handle->header_lock);
  554. }
  555. void starpu_data_partition_submit(starpu_data_handle_t initial_handle, unsigned nparts, starpu_data_handle_t *children)
  556. {
  557. unsigned i;
  558. STARPU_ASSERT_MSG(initial_handle->sequential_consistency, "partition planning is currently only supported for data with sequential consistency");
  559. _starpu_spin_lock(&initial_handle->header_lock);
  560. STARPU_ASSERT_MSG(initial_handle->partitioned == 0, "One can't submit several partition plannings at the same time");
  561. STARPU_ASSERT_MSG(initial_handle->readonly == 0, "One can't submit a partition planning while a readonly partitioning is active");
  562. initial_handle->partitioned++;
  563. initial_handle->active_children = children[0]->siblings;
  564. _starpu_spin_unlock(&initial_handle->header_lock);
  565. for (i = 0; i < nparts; i++)
  566. {
  567. _starpu_spin_lock(&children[i]->header_lock);
  568. children[i]->active = 1;
  569. _starpu_spin_unlock(&children[i]->header_lock);
  570. }
  571. if (!initial_handle->initialized)
  572. /* No need for coherency, it is not initialized */
  573. return;
  574. struct starpu_data_descr descr[nparts];
  575. for (i = 0; i < nparts; i++)
  576. {
  577. STARPU_ASSERT_MSG(children[i]->father_handle == initial_handle, "children parameter of starpu_data_partition_submit must be the children of the parent parameter");
  578. descr[i].handle = children[i];
  579. descr[i].mode = STARPU_W;
  580. }
  581. /* TODO: assert nparts too */
  582. starpu_task_insert(initial_handle->switch_cl, STARPU_RW, initial_handle, STARPU_DATA_MODE_ARRAY, descr, nparts, 0);
  583. starpu_data_invalidate_submit(initial_handle);
  584. }
  585. void starpu_data_partition_readonly_submit(starpu_data_handle_t initial_handle, unsigned nparts, starpu_data_handle_t *children)
  586. {
  587. unsigned i;
  588. STARPU_ASSERT_MSG(initial_handle->sequential_consistency, "partition planning is currently only supported for data with sequential consistency");
  589. _starpu_spin_lock(&initial_handle->header_lock);
  590. STARPU_ASSERT_MSG(initial_handle->partitioned == 0 || initial_handle->readonly, "One can't submit a readonly partition planning at the same time as a readwrite partition planning");
  591. initial_handle->partitioned++;
  592. initial_handle->readonly = 1;
  593. if (initial_handle->nactive_readonly_children < initial_handle->partitioned)
  594. {
  595. _STARPU_REALLOC(initial_handle->active_readonly_children, initial_handle->partitioned * sizeof(initial_handle->active_readonly_children[0]));
  596. initial_handle->nactive_readonly_children = initial_handle->partitioned;
  597. }
  598. initial_handle->active_readonly_children[initial_handle->partitioned-1] = children[0]->siblings;
  599. _starpu_spin_unlock(&initial_handle->header_lock);
  600. for (i = 0; i < nparts; i++)
  601. {
  602. _starpu_spin_lock(&children[i]->header_lock);
  603. children[i]->active = 1;
  604. children[i]->active_ro = 1;
  605. _starpu_spin_unlock(&children[i]->header_lock);
  606. }
  607. STARPU_ASSERT_MSG(initial_handle->initialized, "It is odd to read-only-partition a data which does not have a value yet");
  608. struct starpu_data_descr descr[nparts];
  609. for (i = 0; i < nparts; i++)
  610. {
  611. STARPU_ASSERT_MSG(children[i]->father_handle == initial_handle, "children parameter of starpu_data_partition_submit must be the children of the parent parameter");
  612. descr[i].handle = children[i];
  613. descr[i].mode = STARPU_W;
  614. }
  615. /* TODO: assert nparts too */
  616. starpu_task_insert(initial_handle->switch_cl, STARPU_R, initial_handle, STARPU_DATA_MODE_ARRAY, descr, nparts, 0);
  617. }
  618. void starpu_data_partition_readwrite_upgrade_submit(starpu_data_handle_t initial_handle, unsigned nparts, starpu_data_handle_t *children)
  619. {
  620. STARPU_ASSERT_MSG(initial_handle->sequential_consistency, "partition planning is currently only supported for data with sequential consistency");
  621. _starpu_spin_lock(&initial_handle->header_lock);
  622. STARPU_ASSERT_MSG(initial_handle->partitioned == 1, "One can't upgrade a readonly partition planning to readwrite while other readonly partition plannings are active");
  623. STARPU_ASSERT_MSG(initial_handle->readonly == 1, "One can only upgrade a readonly partition planning");
  624. initial_handle->readonly = 0;
  625. initial_handle->active_children = initial_handle->active_readonly_children[0];
  626. initial_handle->active_readonly_children[0] = NULL;
  627. _starpu_spin_unlock(&initial_handle->header_lock);
  628. unsigned i;
  629. struct starpu_data_descr descr[nparts];
  630. for (i = 0; i < nparts; i++)
  631. {
  632. STARPU_ASSERT_MSG(children[i]->father_handle == initial_handle, "children parameter of starpu_data_partition_submit must be the children of the parent parameter");
  633. children[i]->active_ro = 0;
  634. descr[i].handle = children[i];
  635. descr[i].mode = STARPU_W;
  636. }
  637. /* TODO: assert nparts too */
  638. starpu_task_insert(initial_handle->switch_cl, STARPU_RW, initial_handle, STARPU_DATA_MODE_ARRAY, descr, nparts, 0);
  639. starpu_data_invalidate_submit(initial_handle);
  640. }
  641. void starpu_data_unpartition_submit(starpu_data_handle_t initial_handle, unsigned nparts, starpu_data_handle_t *children, int gather_node)
  642. {
  643. unsigned i;
  644. STARPU_ASSERT_MSG(initial_handle->sequential_consistency, "partition planning is currently only supported for data with sequential consistency");
  645. STARPU_ASSERT_MSG(gather_node == initial_handle->home_node || gather_node == -1, "gathering node different from home node is currently not supported");
  646. _starpu_spin_lock(&initial_handle->header_lock);
  647. STARPU_ASSERT_MSG(initial_handle->partitioned >= 1, "No partition planning is active for this handle");
  648. if (initial_handle->readonly)
  649. {
  650. /* Replace this children set with the last set in the list of readonly children sets */
  651. for (i = 0; i < initial_handle->partitioned-1; i++)
  652. {
  653. if (initial_handle->active_readonly_children[i] == children[0]->siblings)
  654. {
  655. initial_handle->active_readonly_children[i] = initial_handle->active_readonly_children[initial_handle->partitioned-1];
  656. initial_handle->active_readonly_children[initial_handle->partitioned-1] = NULL;
  657. break;
  658. }
  659. }
  660. }
  661. else
  662. {
  663. initial_handle->active_children = NULL;
  664. }
  665. initial_handle->partitioned--;
  666. if (!initial_handle->partitioned)
  667. initial_handle->readonly = 0;
  668. initial_handle->active_children = NULL;
  669. _starpu_spin_unlock(&initial_handle->header_lock);
  670. for (i = 0; i < nparts; i++)
  671. {
  672. _starpu_spin_lock(&children[i]->header_lock);
  673. children[i]->active = 0;
  674. children[i]->active_ro = 0;
  675. _starpu_spin_unlock(&children[i]->header_lock);
  676. }
  677. unsigned n;
  678. struct starpu_data_descr descr[nparts];
  679. for (i = 0, n = 0; i < nparts; i++)
  680. {
  681. STARPU_ASSERT_MSG(children[i]->father_handle == initial_handle, "children parameter of starpu_data_partition_submit must be the children of the parent parameter");
  682. if (!children[i]->initialized)
  683. /* Dropped value, do not care about coherency for this one */
  684. continue;
  685. descr[n].handle = children[i];
  686. descr[n].mode = STARPU_RW;
  687. n++;
  688. }
  689. /* TODO: assert nparts too */
  690. starpu_task_insert(initial_handle->switch_cl, STARPU_W, initial_handle, STARPU_DATA_MODE_ARRAY, descr, n, 0);
  691. for (i = 0; i < nparts; i++)
  692. starpu_data_invalidate_submit(children[i]);
  693. }
  694. void starpu_data_unpartition_readonly_submit(starpu_data_handle_t initial_handle, unsigned nparts, starpu_data_handle_t *children, int gather_node)
  695. {
  696. STARPU_ASSERT_MSG(initial_handle->sequential_consistency, "partition planning is currently only supported for data with sequential consistency");
  697. STARPU_ASSERT_MSG(gather_node == initial_handle->home_node || gather_node == -1, "gathering node different from home node is currently not supported");
  698. _starpu_spin_lock(&initial_handle->header_lock);
  699. STARPU_ASSERT_MSG(initial_handle->partitioned >= 1, "No partition planning is active for this handle");
  700. initial_handle->readonly = 1;
  701. _starpu_spin_unlock(&initial_handle->header_lock);
  702. unsigned i, n;
  703. struct starpu_data_descr descr[nparts];
  704. for (i = 0, n = 0; i < nparts; i++)
  705. {
  706. STARPU_ASSERT_MSG(children[i]->father_handle == initial_handle, "children parameter of starpu_data_partition_submit must be the children of the parent parameter");
  707. if (!children[i]->initialized)
  708. /* Dropped value, do not care about coherency for this one */
  709. continue;
  710. descr[n].handle = children[i];
  711. descr[n].mode = STARPU_R;
  712. n++;
  713. }
  714. /* TODO: assert nparts too */
  715. starpu_task_insert(initial_handle->switch_cl, STARPU_W, initial_handle, STARPU_DATA_MODE_ARRAY, descr, n, 0);
  716. }
  717. /* Unpartition everything below ancestor */
  718. void starpu_data_unpartition_submit_r(starpu_data_handle_t ancestor, int gathering_node)
  719. {
  720. unsigned i, j, nsiblings;
  721. if (!ancestor->partitioned)
  722. /* It's already unpartitioned */
  723. return;
  724. _STARPU_DEBUG("ancestor %p needs unpartitioning\n", ancestor);
  725. if (ancestor->readonly)
  726. {
  727. unsigned n = ancestor->partitioned;
  728. /* Uh, has to go through all read-only partitions */
  729. for (i = 0; i < n; i++)
  730. {
  731. /* Note: active_readonly_children is emptied by starpu_data_unpartition_submit calls */
  732. starpu_data_handle_t *children = ancestor->active_readonly_children[0];
  733. _STARPU_DEBUG("unpartition readonly children %p etc.\n", children[0]);
  734. nsiblings = children[0]->nsiblings;
  735. for (j = 0; j < nsiblings; j++)
  736. {
  737. /* Make sure our children are unpartitioned */
  738. starpu_data_unpartition_submit_r(children[j], gathering_node);
  739. }
  740. /* And unpartition them */
  741. starpu_data_unpartition_submit(ancestor, nsiblings, children, gathering_node);
  742. }
  743. }
  744. else
  745. {
  746. _STARPU_DEBUG("unpartition children %p\n", ancestor->active_children);
  747. /* Only one partition */
  748. nsiblings = ancestor->active_children[0]->nsiblings;
  749. for (i = 0; i < nsiblings; i++)
  750. starpu_data_unpartition_submit_r(ancestor->active_children[i], gathering_node);
  751. /* And unpartition ourself */
  752. starpu_data_unpartition_submit(ancestor, nsiblings, ancestor->active_children, gathering_node);
  753. }
  754. }
  755. /* Make ancestor partition itself properly for target */
  756. static void _starpu_data_partition_access_look_up(starpu_data_handle_t ancestor, starpu_data_handle_t target, int write)
  757. {
  758. /* First make sure ancestor has proper state, if not, ask father */
  759. if (!ancestor->active || (write && ancestor->active_ro))
  760. {
  761. /* (The root is always active-rw) */
  762. STARPU_ASSERT(ancestor->father_handle);
  763. _STARPU_DEBUG("ancestor %p is not ready: %s, asking father %p\n", ancestor, ancestor->active ? ancestor->active_ro ? "RO" : "RW" : "NONE", ancestor->father_handle);
  764. _starpu_data_partition_access_look_up(ancestor->father_handle, ancestor, write);
  765. _STARPU_DEBUG("ancestor %p is now ready\n", ancestor);
  766. }
  767. else
  768. _STARPU_DEBUG("ancestor %p was ready\n", ancestor);
  769. /* We shouldn't be called for nothing */
  770. STARPU_ASSERT(!ancestor->partitioned || !target || ancestor->active_children != target->siblings || (ancestor->readonly && write));
  771. /* Then unpartition ancestor if needed */
  772. if (ancestor->partitioned &&
  773. /* Not the right children, unpartition ourself */
  774. ((target && write && ancestor->active_children != target->siblings) ||
  775. (target && !write && !ancestor->readonly) ||
  776. /* We are partitioned and we want to write or some child
  777. * is writing and we want to read, unpartition ourself*/
  778. (!target && (write || !ancestor->readonly))))
  779. {
  780. #ifdef STARPU_DEVEL
  781. #warning FIXME: better choose gathering node
  782. #endif
  783. starpu_data_unpartition_submit_r(ancestor, STARPU_MAIN_RAM);
  784. }
  785. if (!target)
  786. {
  787. _STARPU_DEBUG("ancestor %p is done\n", ancestor);
  788. /* No child target, nothing more to do actually. */
  789. return;
  790. }
  791. /* Then partition ancestor towards target, if needed */
  792. if (ancestor->partitioned)
  793. {
  794. /* That must be readonly, otherwise we would have unpartitioned it */
  795. STARPU_ASSERT(ancestor->readonly);
  796. if (write)
  797. {
  798. _STARPU_DEBUG("ancestor %p is already partitioned RO, turn RW\n", ancestor);
  799. /* Already partitioned, normally it's already for the target */
  800. STARPU_ASSERT(ancestor->active_children == target->siblings);
  801. /* And we are here just because we haven't partitioned rw */
  802. STARPU_ASSERT(ancestor->readonly && write);
  803. /* So we just need to upgrade ro to rw */
  804. starpu_data_partition_readwrite_upgrade_submit(ancestor, target->nsiblings, target->siblings);
  805. }
  806. else
  807. {
  808. _STARPU_DEBUG("ancestor %p is already partitioned RO, but not to target, partition towards target too\n", ancestor);
  809. /* So we just need to upgrade ro to rw */
  810. starpu_data_partition_readonly_submit(ancestor, target->nsiblings, target->siblings);
  811. }
  812. }
  813. else
  814. {
  815. /* Just need to partition properly for the child */
  816. if (write)
  817. {
  818. _STARPU_DEBUG("partition ancestor %p RW\n", ancestor);
  819. starpu_data_partition_submit(ancestor, target->nsiblings, target->siblings);
  820. }
  821. else
  822. {
  823. _STARPU_DEBUG("partition ancestor %p RO\n", ancestor);
  824. starpu_data_partition_readonly_submit(ancestor, target->nsiblings, target->siblings);
  825. }
  826. }
  827. }
  828. void _starpu_data_partition_access_submit(starpu_data_handle_t target, int write)
  829. {
  830. _STARPU_DEBUG("accessing %p %s\n", target, write ? "RW" : "RO");
  831. _starpu_data_partition_access_look_up(target, NULL, write);
  832. }
  833. /*
  834. * Given an integer N, NPARTS the number of parts it must be divided in, ID the
  835. * part currently considered, determines the CHUNK_SIZE and the OFFSET, taking
  836. * into account the size of the elements stored in the data structure ELEMSIZE
  837. * and LD, the leading dimension.
  838. */
  839. void
  840. _starpu_filter_nparts_compute_chunk_size_and_offset(unsigned n, unsigned nparts,
  841. size_t elemsize, unsigned id,
  842. unsigned ld, unsigned *chunk_size,
  843. size_t *offset)
  844. {
  845. *chunk_size = n/nparts;
  846. unsigned remainder = n % nparts;
  847. if (id < remainder)
  848. (*chunk_size)++;
  849. /*
  850. * Computing the total offset. The formula may not be really clear, but
  851. * it really just is:
  852. *
  853. * total = 0;
  854. * for (i = 0; i < id; i++)
  855. * {
  856. * total += n/nparts;
  857. * if (i < n%nparts)
  858. * total++;
  859. * }
  860. * offset = total * elemsize * ld;
  861. */
  862. if (offset != NULL)
  863. *offset = (id *(n/nparts) + STARPU_MIN(remainder, id)) * ld * elemsize;
  864. }