user_interactions.c 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2009-2017 Université de Bordeaux
  4. * Copyright (C) 2010, 2011, 2012, 2013, 2015, 2016 CNRS
  5. *
  6. * StarPU is free software; you can redistribute it and/or modify
  7. * it under the terms of the GNU Lesser General Public License as published by
  8. * the Free Software Foundation; either version 2.1 of the License, or (at
  9. * your option) any later version.
  10. *
  11. * StarPU is distributed in the hope that it will be useful, but
  12. * WITHOUT ANY WARRANTY; without even the implied warranty of
  13. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  14. *
  15. * See the GNU Lesser General Public License in COPYING.LGPL for more details.
  16. */
  17. #include <common/config.h>
  18. #include <common/utils.h>
  19. #include <core/task.h>
  20. #include <datawizard/coherency.h>
  21. #include <datawizard/copy_driver.h>
  22. #include <datawizard/write_back.h>
  23. #include <core/dependencies/data_concurrency.h>
  24. #include <core/sched_policy.h>
  25. /* Explicitly ask StarPU to allocate room for a piece of data on the specified
  26. * memory node. */
  27. int starpu_data_request_allocation(starpu_data_handle_t handle, unsigned node)
  28. {
  29. struct _starpu_data_request *r;
  30. STARPU_ASSERT(handle);
  31. _starpu_spin_lock(&handle->header_lock);
  32. r = _starpu_create_data_request(handle, NULL, &handle->per_node[node], node, STARPU_NONE, 0, 1, 0, 0, "starpu_data_request_allocation");
  33. /* we do not increase the refcnt associated to the request since we are
  34. * not waiting for its termination */
  35. _starpu_post_data_request(r);
  36. _starpu_spin_unlock(&handle->header_lock);
  37. return 0;
  38. }
  39. struct user_interaction_wrapper
  40. {
  41. starpu_data_handle_t handle;
  42. enum starpu_data_access_mode mode;
  43. int node;
  44. starpu_pthread_cond_t cond;
  45. starpu_pthread_mutex_t lock;
  46. unsigned finished;
  47. unsigned detached;
  48. unsigned prefetch;
  49. unsigned async;
  50. int prio;
  51. void (*callback)(void *);
  52. void (*callback_fetch_data)(void *); // called after fetch_data
  53. void *callback_arg;
  54. struct starpu_task *pre_sync_task;
  55. struct starpu_task *post_sync_task;
  56. };
  57. static inline void _starpu_data_acquire_wrapper_init(struct user_interaction_wrapper *wrapper, starpu_data_handle_t handle, int node, enum starpu_data_access_mode mode)
  58. {
  59. memset(wrapper, 0, sizeof(*wrapper));
  60. wrapper->handle = handle;
  61. wrapper->node = node;
  62. wrapper->mode = mode;
  63. wrapper->finished = 0;
  64. STARPU_PTHREAD_COND_INIT(&wrapper->cond, NULL);
  65. STARPU_PTHREAD_MUTEX_INIT(&wrapper->lock, NULL);
  66. }
  67. /* Called to signal completion of asynchronous data acquisition */
  68. static inline void _starpu_data_acquire_wrapper_finished(struct user_interaction_wrapper *wrapper)
  69. {
  70. STARPU_PTHREAD_MUTEX_LOCK(&wrapper->lock);
  71. wrapper->finished = 1;
  72. STARPU_PTHREAD_COND_SIGNAL(&wrapper->cond);
  73. STARPU_PTHREAD_MUTEX_UNLOCK(&wrapper->lock);
  74. }
  75. /* Called to wait for completion of asynchronous data acquisition */
  76. static inline void _starpu_data_acquire_wrapper_wait(struct user_interaction_wrapper *wrapper)
  77. {
  78. STARPU_PTHREAD_MUTEX_LOCK(&wrapper->lock);
  79. while (!wrapper->finished)
  80. STARPU_PTHREAD_COND_WAIT(&wrapper->cond, &wrapper->lock);
  81. STARPU_PTHREAD_MUTEX_UNLOCK(&wrapper->lock);
  82. }
  83. static inline void _starpu_data_acquire_wrapper_fini(struct user_interaction_wrapper *wrapper)
  84. {
  85. STARPU_PTHREAD_COND_DESTROY(&wrapper->cond);
  86. STARPU_PTHREAD_MUTEX_DESTROY(&wrapper->lock);
  87. }
  88. /* Called when the fetch into target memory is done, we're done! */
  89. static inline void _starpu_data_acquire_fetch_done(struct user_interaction_wrapper *wrapper)
  90. {
  91. if (wrapper->node >= 0)
  92. {
  93. struct _starpu_data_replicate *replicate = &wrapper->handle->per_node[wrapper->node];
  94. if (replicate->mc)
  95. replicate->mc->diduse = 1;
  96. }
  97. }
  98. /* Called when the data acquisition is done, to launch the fetch into target memory */
  99. static inline void _starpu_data_acquire_launch_fetch(struct user_interaction_wrapper *wrapper, int async, void (*callback)(void *), void *callback_arg)
  100. {
  101. int node = wrapper->node;
  102. starpu_data_handle_t handle = wrapper->handle;
  103. struct _starpu_data_replicate *replicate = node >= 0 ? &handle->per_node[node] : NULL;
  104. int ret = _starpu_fetch_data_on_node(handle, node, replicate, wrapper->mode, wrapper->detached, wrapper->prefetch, async, callback, callback_arg, wrapper->prio, "_starpu_data_acquire_launch_fetch");
  105. STARPU_ASSERT(!ret);
  106. }
  107. /*
  108. * Non Blocking data request from application
  109. */
  110. /* Called when fetch is done, call the callback */
  111. static void _starpu_data_acquire_fetch_data_callback(void *arg)
  112. {
  113. struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) arg;
  114. starpu_data_handle_t handle = wrapper->handle;
  115. /* At that moment, the caller holds a reference to the piece of data.
  116. * We enqueue the "post" sync task in the list associated to the handle
  117. * so that it is submitted by the starpu_data_release
  118. * function. */
  119. if (wrapper->post_sync_task)
  120. _starpu_add_post_sync_tasks(wrapper->post_sync_task, handle);
  121. _starpu_data_acquire_fetch_done(wrapper);
  122. wrapper->callback(wrapper->callback_arg);
  123. _starpu_data_acquire_wrapper_fini(wrapper);
  124. free(wrapper);
  125. }
  126. /* Called when the data acquisition is done, launch the fetch into target memory */
  127. static void _starpu_data_acquire_continuation_non_blocking(void *arg)
  128. {
  129. _starpu_data_acquire_launch_fetch(arg, 1, _starpu_data_acquire_fetch_data_callback, arg);
  130. }
  131. /* Called when the implicit data dependencies are done, launch the data acquisition */
  132. static void starpu_data_acquire_cb_pre_sync_callback(void *arg)
  133. {
  134. struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) arg;
  135. /* we try to get the data, if we do not succeed immediately, we set a
  136. * callback function that will be executed automatically when the data is
  137. * available again, otherwise we fetch the data directly */
  138. if (!_starpu_attempt_to_submit_data_request_from_apps(wrapper->handle, wrapper->mode,
  139. _starpu_data_acquire_continuation_non_blocking, wrapper))
  140. {
  141. /* no one has locked this data yet, so we proceed immediately */
  142. _starpu_data_acquire_continuation_non_blocking(wrapper);
  143. }
  144. }
  145. /* The data must be released by calling starpu_data_release later on */
  146. int starpu_data_acquire_on_node_cb_sequential_consistency_sync_jobids(starpu_data_handle_t handle, int node,
  147. enum starpu_data_access_mode mode, void (*callback)(void *), void *arg,
  148. int sequential_consistency,
  149. long *pre_sync_jobid, long *post_sync_jobid)
  150. {
  151. STARPU_ASSERT(handle);
  152. STARPU_ASSERT_MSG(handle->nchildren == 0, "Acquiring a partitioned data (%p) is not possible", handle);
  153. _STARPU_LOG_IN();
  154. struct user_interaction_wrapper *wrapper;
  155. _STARPU_MALLOC(wrapper, sizeof(struct user_interaction_wrapper));
  156. _starpu_data_acquire_wrapper_init(wrapper, handle, node, mode);
  157. wrapper->async = 1;
  158. wrapper->callback = callback;
  159. wrapper->callback_arg = arg;
  160. wrapper->pre_sync_task = NULL;
  161. wrapper->post_sync_task = NULL;
  162. STARPU_PTHREAD_MUTEX_LOCK(&handle->sequential_consistency_mutex);
  163. int handle_sequential_consistency = handle->sequential_consistency;
  164. if (handle_sequential_consistency && sequential_consistency)
  165. {
  166. struct starpu_task *new_task;
  167. wrapper->pre_sync_task = starpu_task_create();
  168. wrapper->pre_sync_task->name = "_starpu_data_acquire_cb_pre";
  169. wrapper->pre_sync_task->detach = 1;
  170. wrapper->pre_sync_task->callback_func = starpu_data_acquire_cb_pre_sync_callback;
  171. wrapper->pre_sync_task->callback_arg = wrapper;
  172. if (pre_sync_jobid)
  173. *pre_sync_jobid = _starpu_get_job_associated_to_task(wrapper->pre_sync_task)->job_id;
  174. wrapper->post_sync_task = starpu_task_create();
  175. wrapper->post_sync_task->name = "_starpu_data_acquire_cb_post";
  176. wrapper->post_sync_task->detach = 1;
  177. if (post_sync_jobid)
  178. *post_sync_jobid = _starpu_get_job_associated_to_task(wrapper->post_sync_task)->job_id;
  179. new_task = _starpu_detect_implicit_data_deps_with_handle(wrapper->pre_sync_task, wrapper->post_sync_task, &_starpu_get_job_associated_to_task(wrapper->post_sync_task)->implicit_dep_slot, handle, mode);
  180. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->sequential_consistency_mutex);
  181. if (new_task)
  182. {
  183. int ret = _starpu_task_submit_internally(new_task);
  184. STARPU_ASSERT(!ret);
  185. }
  186. /* TODO detect if this is superflous */
  187. int ret = _starpu_task_submit_internally(wrapper->pre_sync_task);
  188. STARPU_ASSERT(!ret);
  189. }
  190. else
  191. {
  192. if (pre_sync_jobid)
  193. *pre_sync_jobid = -1;
  194. if (post_sync_jobid)
  195. *post_sync_jobid = -1;
  196. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->sequential_consistency_mutex);
  197. starpu_data_acquire_cb_pre_sync_callback(wrapper);
  198. }
  199. _STARPU_LOG_OUT();
  200. return 0;
  201. }
  202. int starpu_data_acquire_on_node_cb_sequential_consistency(starpu_data_handle_t handle, int node,
  203. enum starpu_data_access_mode mode, void (*callback)(void *), void *arg,
  204. int sequential_consistency)
  205. {
  206. return starpu_data_acquire_on_node_cb_sequential_consistency_sync_jobids(handle, node, mode, callback, arg, sequential_consistency, NULL, NULL);
  207. }
  208. int starpu_data_acquire_on_node_cb(starpu_data_handle_t handle, int node,
  209. enum starpu_data_access_mode mode, void (*callback)(void *), void *arg)
  210. {
  211. return starpu_data_acquire_on_node_cb_sequential_consistency(handle, node, mode, callback, arg, 1);
  212. }
  213. int starpu_data_acquire_cb(starpu_data_handle_t handle,
  214. enum starpu_data_access_mode mode, void (*callback)(void *), void *arg)
  215. {
  216. return starpu_data_acquire_on_node_cb(handle, STARPU_MAIN_RAM, mode, callback, arg);
  217. }
  218. int starpu_data_acquire_cb_sequential_consistency(starpu_data_handle_t handle,
  219. enum starpu_data_access_mode mode, void (*callback)(void *), void *arg, int sequential_consistency)
  220. {
  221. return starpu_data_acquire_on_node_cb_sequential_consistency(handle, STARPU_MAIN_RAM, mode, callback, arg, sequential_consistency);
  222. }
  223. /*
  224. * Blockin data request from application
  225. */
  226. static inline void _starpu_data_acquire_continuation(void *arg)
  227. {
  228. struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) arg;
  229. starpu_data_handle_t handle = wrapper->handle;
  230. STARPU_ASSERT(handle);
  231. _starpu_data_acquire_launch_fetch(wrapper, 0, NULL, NULL);
  232. _starpu_data_acquire_fetch_done(wrapper);
  233. _starpu_data_acquire_wrapper_finished(wrapper);
  234. }
  235. /* The data must be released by calling starpu_data_release later on */
  236. int starpu_data_acquire_on_node(starpu_data_handle_t handle, int node, enum starpu_data_access_mode mode)
  237. {
  238. STARPU_ASSERT(handle);
  239. STARPU_ASSERT_MSG(handle->nchildren == 0, "Acquiring a partitioned data is not possible");
  240. _STARPU_LOG_IN();
  241. /* unless asynchronous, it is forbidden to call this function from a callback or a codelet */
  242. STARPU_ASSERT_MSG(_starpu_worker_may_perform_blocking_calls(), "Acquiring a data synchronously is not possible from a codelet or from a task callback, use starpu_data_acquire_cb instead.");
  243. if (node >= 0 && _starpu_data_is_multiformat_handle(handle) &&
  244. _starpu_handle_needs_conversion_task(handle, node))
  245. {
  246. struct starpu_task *task = _starpu_create_conversion_task(handle, node);
  247. int ret;
  248. _starpu_spin_lock(&handle->header_lock);
  249. handle->refcnt--;
  250. handle->busy_count--;
  251. handle->mf_node = node;
  252. _starpu_spin_unlock(&handle->header_lock);
  253. task->synchronous = 1;
  254. ret = _starpu_task_submit_internally(task);
  255. STARPU_ASSERT(!ret);
  256. }
  257. struct user_interaction_wrapper wrapper;
  258. _starpu_data_acquire_wrapper_init(&wrapper, handle, node, mode);
  259. // _STARPU_DEBUG("TAKE sequential_consistency_mutex starpu_data_acquire\n");
  260. STARPU_PTHREAD_MUTEX_LOCK(&handle->sequential_consistency_mutex);
  261. int sequential_consistency = handle->sequential_consistency;
  262. if (sequential_consistency)
  263. {
  264. struct starpu_task *new_task;
  265. wrapper.pre_sync_task = starpu_task_create();
  266. wrapper.pre_sync_task->name = "_starpu_data_acquire_pre";
  267. wrapper.pre_sync_task->detach = 0;
  268. wrapper.post_sync_task = starpu_task_create();
  269. wrapper.post_sync_task->name = "_starpu_data_acquire_post";
  270. wrapper.post_sync_task->detach = 1;
  271. new_task = _starpu_detect_implicit_data_deps_with_handle(wrapper.pre_sync_task, wrapper.post_sync_task, &_starpu_get_job_associated_to_task(wrapper.post_sync_task)->implicit_dep_slot, handle, mode);
  272. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->sequential_consistency_mutex);
  273. if (new_task)
  274. {
  275. int ret = _starpu_task_submit_internally(new_task);
  276. STARPU_ASSERT(!ret);
  277. }
  278. /* TODO detect if this is superflous */
  279. wrapper.pre_sync_task->synchronous = 1;
  280. int ret = _starpu_task_submit_internally(wrapper.pre_sync_task);
  281. STARPU_ASSERT(!ret);
  282. }
  283. else
  284. {
  285. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->sequential_consistency_mutex);
  286. }
  287. /* we try to get the data, if we do not succeed immediately, we set a
  288. * callback function that will be executed automatically when the data is
  289. * available again, otherwise we fetch the data directly */
  290. if (!_starpu_attempt_to_submit_data_request_from_apps(handle, mode, _starpu_data_acquire_continuation, &wrapper))
  291. {
  292. /* no one has locked this data yet, so we proceed immediately */
  293. _starpu_data_acquire_launch_fetch(&wrapper, 0, NULL, NULL);
  294. _starpu_data_acquire_fetch_done(&wrapper);
  295. }
  296. else
  297. {
  298. _starpu_data_acquire_wrapper_wait(&wrapper);
  299. }
  300. _starpu_data_acquire_wrapper_fini(&wrapper);
  301. /* At that moment, the caller holds a reference to the piece of data.
  302. * We enqueue the "post" sync task in the list associated to the handle
  303. * so that it is submitted by the starpu_data_release
  304. * function. */
  305. if (sequential_consistency)
  306. _starpu_add_post_sync_tasks(wrapper.post_sync_task, handle);
  307. _STARPU_LOG_OUT();
  308. return 0;
  309. }
  310. int starpu_data_acquire(starpu_data_handle_t handle, enum starpu_data_access_mode mode)
  311. {
  312. return starpu_data_acquire_on_node(handle, STARPU_MAIN_RAM, mode);
  313. }
  314. int starpu_data_acquire_on_node_try(starpu_data_handle_t handle, int node, enum starpu_data_access_mode mode)
  315. {
  316. STARPU_ASSERT(handle);
  317. STARPU_ASSERT_MSG(handle->nchildren == 0, "Acquiring a partitioned data is not possible");
  318. /* it is forbidden to call this function from a callback or a codelet */
  319. STARPU_ASSERT_MSG(_starpu_worker_may_perform_blocking_calls(), "Acquiring a data synchronously is not possible from a codelet or from a task callback, use starpu_data_acquire_cb instead.");
  320. int ret;
  321. STARPU_ASSERT_MSG(!_starpu_data_is_multiformat_handle(handle), "not supported yet");
  322. STARPU_PTHREAD_MUTEX_LOCK(&handle->sequential_consistency_mutex);
  323. ret = _starpu_test_implicit_data_deps_with_handle(handle, mode);
  324. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->sequential_consistency_mutex);
  325. if (ret)
  326. return ret;
  327. struct user_interaction_wrapper wrapper;
  328. _starpu_data_acquire_wrapper_init(&wrapper, handle, node, mode);
  329. /* we try to get the data, if we do not succeed immediately, we set a
  330. * callback function that will be executed automatically when the data is
  331. * available again, otherwise we fetch the data directly */
  332. if (!_starpu_attempt_to_submit_data_request_from_apps(handle, mode, _starpu_data_acquire_continuation, &wrapper))
  333. {
  334. /* no one has locked this data yet, so we proceed immediately */
  335. _starpu_data_acquire_launch_fetch(&wrapper, 0, NULL, NULL);
  336. _starpu_data_acquire_fetch_done(&wrapper);
  337. }
  338. else
  339. {
  340. _starpu_data_acquire_wrapper_wait(&wrapper);
  341. }
  342. _starpu_data_acquire_wrapper_fini(&wrapper);
  343. return 0;
  344. }
  345. int starpu_data_acquire_try(starpu_data_handle_t handle, enum starpu_data_access_mode mode)
  346. {
  347. return starpu_data_acquire_on_node_try(handle, STARPU_MAIN_RAM, mode);
  348. }
  349. /* This function must be called after starpu_data_acquire so that the
  350. * application release the data */
  351. void starpu_data_release_on_node(starpu_data_handle_t handle, int node)
  352. {
  353. STARPU_ASSERT(handle);
  354. /* In case there are some implicit dependencies, unlock the "post sync" tasks */
  355. _starpu_unlock_post_sync_tasks(handle);
  356. /* The application can now release the rw-lock */
  357. if (node >= 0)
  358. _starpu_release_data_on_node(handle, 0, &handle->per_node[node]);
  359. else
  360. {
  361. _starpu_spin_lock(&handle->header_lock);
  362. if (node == STARPU_ACQUIRE_NO_NODE_LOCK_ALL)
  363. {
  364. int i;
  365. for (i = 0; i < STARPU_MAXNODES; i++)
  366. handle->per_node[i].refcnt--;
  367. }
  368. handle->busy_count--;
  369. if (!_starpu_notify_data_dependencies(handle))
  370. _starpu_spin_unlock(&handle->header_lock);
  371. }
  372. }
  373. void starpu_data_release(starpu_data_handle_t handle)
  374. {
  375. starpu_data_release_on_node(handle, STARPU_MAIN_RAM);
  376. }
  377. static void _prefetch_data_on_node(void *arg)
  378. {
  379. struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) arg;
  380. starpu_data_handle_t handle = wrapper->handle;
  381. _starpu_data_acquire_launch_fetch(wrapper, wrapper->async, NULL, NULL);
  382. if (wrapper->async)
  383. free(wrapper);
  384. else
  385. _starpu_data_acquire_wrapper_finished(wrapper);
  386. _starpu_spin_lock(&handle->header_lock);
  387. if (!_starpu_notify_data_dependencies(handle))
  388. _starpu_spin_unlock(&handle->header_lock);
  389. }
  390. static
  391. int _starpu_prefetch_data_on_node_with_mode(starpu_data_handle_t handle, unsigned node, unsigned async, enum starpu_data_access_mode mode, unsigned prefetch, int prio)
  392. {
  393. STARPU_ASSERT(handle);
  394. /* it is forbidden to call this function from a callback or a codelet */
  395. STARPU_ASSERT_MSG(async || _starpu_worker_may_perform_blocking_calls(), "Synchronous prefetch is not possible from a task or a callback");
  396. struct user_interaction_wrapper *wrapper;
  397. _STARPU_MALLOC(wrapper, sizeof(*wrapper));
  398. _starpu_data_acquire_wrapper_init(wrapper, handle, node, STARPU_R);
  399. wrapper->detached = async;
  400. wrapper->prefetch = prefetch;
  401. wrapper->async = async;
  402. wrapper->prio = prio;
  403. if (!_starpu_attempt_to_submit_data_request_from_apps(handle, mode, _prefetch_data_on_node, wrapper))
  404. {
  405. /* we can immediately proceed */
  406. struct _starpu_data_replicate *replicate = &handle->per_node[node];
  407. _starpu_data_acquire_launch_fetch(wrapper, async, NULL, NULL);
  408. _starpu_data_acquire_wrapper_fini(wrapper);
  409. free(wrapper);
  410. /* remove the "lock"/reference */
  411. _starpu_spin_lock(&handle->header_lock);
  412. if (!async)
  413. {
  414. /* Release our refcnt, like _starpu_release_data_on_node would do */
  415. replicate->refcnt--;
  416. STARPU_ASSERT(replicate->refcnt >= 0);
  417. STARPU_ASSERT(handle->busy_count > 0);
  418. handle->busy_count--;
  419. }
  420. /* In case there was a temporary handle (eg. used for reduction), this
  421. * handle may have requested to be destroyed when the data is released
  422. * */
  423. if (!_starpu_notify_data_dependencies(handle))
  424. _starpu_spin_unlock(&handle->header_lock);
  425. }
  426. else if (!async)
  427. {
  428. _starpu_data_acquire_wrapper_wait(wrapper);
  429. _starpu_data_acquire_wrapper_fini(wrapper);
  430. free(wrapper);
  431. }
  432. return 0;
  433. }
  434. int starpu_data_fetch_on_node(starpu_data_handle_t handle, unsigned node, unsigned async)
  435. {
  436. return _starpu_prefetch_data_on_node_with_mode(handle, node, async, STARPU_R, 0, 0);
  437. }
  438. int starpu_data_prefetch_on_node_prio(starpu_data_handle_t handle, unsigned node, unsigned async, int prio)
  439. {
  440. return _starpu_prefetch_data_on_node_with_mode(handle, node, async, STARPU_R, 1, prio);
  441. }
  442. int starpu_data_prefetch_on_node(starpu_data_handle_t handle, unsigned node, unsigned async)
  443. {
  444. return starpu_data_prefetch_on_node_prio(handle, node, async, 0);
  445. }
  446. int starpu_data_idle_prefetch_on_node_prio(starpu_data_handle_t handle, unsigned node, unsigned async, int prio)
  447. {
  448. return _starpu_prefetch_data_on_node_with_mode(handle, node, async, STARPU_R, 2, prio);
  449. }
  450. int starpu_data_idle_prefetch_on_node(starpu_data_handle_t handle, unsigned node, unsigned async)
  451. {
  452. return starpu_data_idle_prefetch_on_node_prio(handle, node, async, 0);
  453. }
  454. static void _starpu_data_wont_use(void *data)
  455. {
  456. unsigned node;
  457. starpu_data_handle_t handle = data;
  458. _starpu_spin_lock(&handle->header_lock);
  459. for (node = 0; node < STARPU_MAXNODES; node++)
  460. {
  461. struct _starpu_data_replicate *local = &handle->per_node[node];
  462. if (local->allocated && local->automatically_allocated)
  463. _starpu_memchunk_wont_use(local->mc, node);
  464. }
  465. if (handle->per_worker)
  466. {
  467. unsigned nworkers = starpu_worker_get_count();
  468. unsigned worker;
  469. for (worker = 0; worker < nworkers; worker++)
  470. {
  471. struct _starpu_data_replicate *local = &handle->per_worker[worker];
  472. if (local->allocated && local->automatically_allocated)
  473. _starpu_memchunk_wont_use(local->mc, starpu_worker_get_memory_node(worker));
  474. }
  475. }
  476. _starpu_spin_unlock(&handle->header_lock);
  477. starpu_data_release_on_node(handle, STARPU_ACQUIRE_NO_NODE_LOCK_ALL);
  478. if (handle->home_node != -1)
  479. starpu_data_idle_prefetch_on_node(handle, handle->home_node, 1);
  480. }
  481. void starpu_data_wont_use(starpu_data_handle_t handle)
  482. {
  483. starpu_data_acquire_on_node_cb(handle, STARPU_ACQUIRE_NO_NODE_LOCK_ALL, STARPU_R, _starpu_data_wont_use, handle);
  484. }
  485. /*
  486. * It is possible to specify that a piece of data can be discarded without
  487. * impacting the application.
  488. */
  489. int _starpu_has_not_important_data;
  490. void starpu_data_advise_as_important(starpu_data_handle_t handle, unsigned is_important)
  491. {
  492. if (!is_important)
  493. _starpu_has_not_important_data = 1;
  494. _starpu_spin_lock(&handle->header_lock);
  495. /* first take all the children lock (in order !) */
  496. unsigned child;
  497. for (child = 0; child < handle->nchildren; child++)
  498. {
  499. /* make sure the intermediate children is advised as well */
  500. starpu_data_handle_t child_handle = starpu_data_get_child(handle, child);
  501. if (child_handle->nchildren > 0)
  502. starpu_data_advise_as_important(child_handle, is_important);
  503. }
  504. handle->is_not_important = !is_important;
  505. /* now the parent may be used again so we release the lock */
  506. _starpu_spin_unlock(&handle->header_lock);
  507. }
  508. void starpu_data_set_sequential_consistency_flag(starpu_data_handle_t handle, unsigned flag)
  509. {
  510. _starpu_spin_lock(&handle->header_lock);
  511. unsigned child;
  512. for (child = 0; child < handle->nchildren; child++)
  513. {
  514. /* make sure that the flags are applied to the children as well */
  515. starpu_data_handle_t child_handle = starpu_data_get_child(handle, child);
  516. if (child_handle->nchildren > 0)
  517. starpu_data_set_sequential_consistency_flag(child_handle, flag);
  518. }
  519. STARPU_PTHREAD_MUTEX_LOCK(&handle->sequential_consistency_mutex);
  520. handle->sequential_consistency = flag;
  521. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->sequential_consistency_mutex);
  522. _starpu_spin_unlock(&handle->header_lock);
  523. }
  524. unsigned starpu_data_get_sequential_consistency_flag(starpu_data_handle_t handle)
  525. {
  526. return handle->sequential_consistency;
  527. }
  528. /* By default, sequential consistency is enabled */
  529. static unsigned default_sequential_consistency_flag = 1;
  530. unsigned starpu_data_get_default_sequential_consistency_flag(void)
  531. {
  532. return default_sequential_consistency_flag;
  533. }
  534. void starpu_data_set_default_sequential_consistency_flag(unsigned flag)
  535. {
  536. default_sequential_consistency_flag = flag;
  537. }
  538. /* Query the status of the handle on the specified memory node. */
  539. void starpu_data_query_status(starpu_data_handle_t handle, int memory_node, int *is_allocated, int *is_valid, int *is_requested)
  540. {
  541. // XXX : this is just a hint, so we don't take the lock ...
  542. // _starpu_spin_lock(&handle->header_lock);
  543. if (is_allocated)
  544. *is_allocated = handle->per_node[memory_node].allocated;
  545. if (is_valid)
  546. *is_valid = (handle->per_node[memory_node].state != STARPU_INVALID);
  547. if (is_requested)
  548. {
  549. int requested = 0;
  550. unsigned node;
  551. for (node = 0; node < STARPU_MAXNODES; node++)
  552. {
  553. if (handle->per_node[memory_node].requested & (1UL << node))
  554. {
  555. requested = 1;
  556. break;
  557. }
  558. }
  559. *is_requested = requested;
  560. }
  561. // _starpu_spin_unlock(&handle->header_lock);
  562. }