user_interactions.c 25 KB

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