user_interactions.c 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2009-2013 Université de Bordeaux 1
  4. * Copyright (C) 2010, 2011, 2012, 2013 Centre National de la Recherche Scientifique
  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);
  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, node);
  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. unsigned node;
  44. starpu_pthread_cond_t cond;
  45. starpu_pthread_mutex_t lock;
  46. unsigned finished;
  47. unsigned async;
  48. void (*callback)(void *);
  49. void (*callback_fetch_data)(void *); // called after fetch_data
  50. void *callback_arg;
  51. struct starpu_task *pre_sync_task;
  52. struct starpu_task *post_sync_task;
  53. };
  54. /*
  55. * Non Blocking data request from application
  56. */
  57. /* put the current value of the data into RAM */
  58. static void _starpu_data_acquire_fetch_data_callback(void *arg)
  59. {
  60. struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) arg;
  61. starpu_data_handle_t handle = wrapper->handle;
  62. /* At that moment, the caller holds a reference to the piece of data.
  63. * We enqueue the "post" sync task in the list associated to the handle
  64. * so that it is submitted by the starpu_data_release
  65. * function. */
  66. _starpu_add_post_sync_tasks(wrapper->post_sync_task, handle);
  67. wrapper->callback(wrapper->callback_arg);
  68. free(wrapper);
  69. }
  70. static void _starpu_data_acquire_continuation_non_blocking(void *arg)
  71. {
  72. int ret;
  73. struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) arg;
  74. starpu_data_handle_t handle = wrapper->handle;
  75. STARPU_ASSERT(handle);
  76. struct _starpu_data_replicate *replicate = &handle->per_node[wrapper->node];
  77. ret = _starpu_fetch_data_on_node(handle, replicate, wrapper->mode, 0, 1,
  78. _starpu_data_acquire_fetch_data_callback, wrapper);
  79. STARPU_ASSERT(!ret);
  80. }
  81. static void starpu_data_acquire_cb_pre_sync_callback(void *arg)
  82. {
  83. struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) arg;
  84. /* we try to get the data, if we do not succeed immediately, we set a
  85. * callback function that will be executed automatically when the data is
  86. * available again, otherwise we fetch the data directly */
  87. if (!_starpu_attempt_to_submit_data_request_from_apps(wrapper->handle, wrapper->mode,
  88. _starpu_data_acquire_continuation_non_blocking, wrapper))
  89. {
  90. /* no one has locked this data yet, so we proceed immediately */
  91. _starpu_data_acquire_continuation_non_blocking(wrapper);
  92. }
  93. }
  94. /* The data must be released by calling starpu_data_release later on */
  95. int starpu_data_acquire_on_node_cb(starpu_data_handle_t handle, unsigned node,
  96. enum starpu_data_access_mode mode, void (*callback)(void *), void *arg)
  97. {
  98. STARPU_ASSERT(handle);
  99. STARPU_ASSERT_MSG(handle->nchildren == 0, "Acquiring a partitioned data is not possible");
  100. _STARPU_LOG_IN();
  101. struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) malloc(sizeof(struct user_interaction_wrapper));
  102. STARPU_ASSERT(wrapper);
  103. wrapper->handle = handle;
  104. wrapper->node = node;
  105. wrapper->mode = mode;
  106. wrapper->callback = callback;
  107. wrapper->callback_arg = arg;
  108. STARPU_PTHREAD_COND_INIT(&wrapper->cond, NULL);
  109. STARPU_PTHREAD_MUTEX_INIT(&wrapper->lock, NULL);
  110. wrapper->finished = 0;
  111. STARPU_PTHREAD_MUTEX_LOCK(&handle->sequential_consistency_mutex);
  112. int sequential_consistency = handle->sequential_consistency;
  113. if (sequential_consistency)
  114. {
  115. struct starpu_task *new_task;
  116. wrapper->pre_sync_task = starpu_task_create();
  117. wrapper->pre_sync_task->detach = 1;
  118. wrapper->pre_sync_task->callback_func = starpu_data_acquire_cb_pre_sync_callback;
  119. wrapper->pre_sync_task->callback_arg = wrapper;
  120. wrapper->post_sync_task = starpu_task_create();
  121. wrapper->post_sync_task->detach = 1;
  122. #ifdef STARPU_USE_FXT
  123. struct _starpu_job *job = _starpu_get_job_associated_to_task(wrapper->pre_sync_task);
  124. job->model_name = "acquire_cb_pre";
  125. job = _starpu_get_job_associated_to_task(wrapper->post_sync_task);
  126. job->model_name = "acquire_cb_post";
  127. #endif
  128. new_task = _starpu_detect_implicit_data_deps_with_handle(wrapper->pre_sync_task, wrapper->post_sync_task, handle, mode);
  129. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->sequential_consistency_mutex);
  130. if (new_task)
  131. {
  132. int ret = _starpu_task_submit_internally(new_task);
  133. STARPU_ASSERT(!ret);
  134. }
  135. /* TODO detect if this is superflous */
  136. int ret = _starpu_task_submit_internally(wrapper->pre_sync_task);
  137. STARPU_ASSERT(!ret);
  138. }
  139. else
  140. {
  141. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->sequential_consistency_mutex);
  142. starpu_data_acquire_cb_pre_sync_callback(wrapper);
  143. }
  144. _STARPU_LOG_OUT();
  145. return 0;
  146. }
  147. int starpu_data_acquire_cb(starpu_data_handle_t handle,
  148. enum starpu_data_access_mode mode, void (*callback)(void *), void *arg)
  149. {
  150. return starpu_data_acquire_on_node_cb(handle, 0, mode, callback, arg);
  151. }
  152. /*
  153. * Block data request from application
  154. */
  155. static inline void _starpu_data_acquire_continuation(void *arg)
  156. {
  157. struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) arg;
  158. starpu_data_handle_t handle = wrapper->handle;
  159. STARPU_ASSERT(handle);
  160. struct _starpu_data_replicate *replicate = &handle->per_node[wrapper->node];
  161. _starpu_fetch_data_on_node(handle, replicate, wrapper->mode, 0, 0, NULL, NULL);
  162. /* continuation of starpu_data_acquire */
  163. STARPU_PTHREAD_MUTEX_LOCK(&wrapper->lock);
  164. wrapper->finished = 1;
  165. STARPU_PTHREAD_COND_SIGNAL(&wrapper->cond);
  166. STARPU_PTHREAD_MUTEX_UNLOCK(&wrapper->lock);
  167. }
  168. /* The data must be released by calling starpu_data_release later on */
  169. int starpu_data_acquire_on_node(starpu_data_handle_t handle, unsigned node, enum starpu_data_access_mode mode)
  170. {
  171. STARPU_ASSERT(handle);
  172. STARPU_ASSERT_MSG(handle->nchildren == 0, "Acquiring a partitioned data is not possible");
  173. _STARPU_LOG_IN();
  174. /* unless asynchronous, it is forbidden to call this function from a callback or a codelet */
  175. if (STARPU_UNLIKELY(!_starpu_worker_may_perform_blocking_calls()))
  176. {
  177. _STARPU_LOG_OUT_TAG("EDEADLK");
  178. return -EDEADLK;
  179. }
  180. if (_starpu_data_is_multiformat_handle(handle) &&
  181. _starpu_handle_needs_conversion_task(handle, 0))
  182. {
  183. struct starpu_task *task = _starpu_create_conversion_task(handle, 0);
  184. int ret;
  185. _starpu_spin_lock(&handle->header_lock);
  186. handle->refcnt--;
  187. handle->busy_count--;
  188. handle->mf_node = 0;
  189. _starpu_spin_unlock(&handle->header_lock);
  190. task->synchronous = 1;
  191. ret = _starpu_task_submit_internally(task);
  192. STARPU_ASSERT(!ret);
  193. }
  194. struct user_interaction_wrapper wrapper =
  195. {
  196. .handle = handle,
  197. .mode = mode,
  198. .node = node,
  199. .finished = 0
  200. };
  201. STARPU_PTHREAD_COND_INIT(&wrapper.cond, NULL);
  202. STARPU_PTHREAD_MUTEX_INIT(&wrapper.lock, NULL);
  203. // _STARPU_DEBUG("TAKE sequential_consistency_mutex starpu_data_acquire\n");
  204. STARPU_PTHREAD_MUTEX_LOCK(&handle->sequential_consistency_mutex);
  205. int sequential_consistency = handle->sequential_consistency;
  206. if (sequential_consistency)
  207. {
  208. struct starpu_task *new_task;
  209. wrapper.pre_sync_task = starpu_task_create();
  210. wrapper.pre_sync_task->detach = 0;
  211. wrapper.post_sync_task = starpu_task_create();
  212. wrapper.post_sync_task->detach = 1;
  213. #ifdef STARPU_USE_FXT
  214. struct _starpu_job *job = _starpu_get_job_associated_to_task(wrapper.pre_sync_task);
  215. job->model_name = "acquire_pre";
  216. job = _starpu_get_job_associated_to_task(wrapper.post_sync_task);
  217. job->model_name = "acquire_post";
  218. #endif
  219. new_task = _starpu_detect_implicit_data_deps_with_handle(wrapper.pre_sync_task, wrapper.post_sync_task, handle, mode);
  220. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->sequential_consistency_mutex);
  221. if (new_task)
  222. {
  223. int ret = _starpu_task_submit_internally(new_task);
  224. STARPU_ASSERT(!ret);
  225. }
  226. /* TODO detect if this is superflous */
  227. wrapper.pre_sync_task->synchronous = 1;
  228. int ret = _starpu_task_submit_internally(wrapper.pre_sync_task);
  229. STARPU_ASSERT(!ret);
  230. }
  231. else
  232. {
  233. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->sequential_consistency_mutex);
  234. }
  235. /* we try to get the data, if we do not succeed immediately, we set a
  236. * callback function that will be executed automatically when the data is
  237. * available again, otherwise we fetch the data directly */
  238. if (!_starpu_attempt_to_submit_data_request_from_apps(handle, mode, _starpu_data_acquire_continuation, &wrapper))
  239. {
  240. /* no one has locked this data yet, so we proceed immediately */
  241. struct _starpu_data_replicate *replicate = &handle->per_node[node];
  242. int ret = _starpu_fetch_data_on_node(handle, replicate, mode, 0, 0, NULL, NULL);
  243. STARPU_ASSERT(!ret);
  244. }
  245. else
  246. {
  247. STARPU_PTHREAD_MUTEX_LOCK(&wrapper.lock);
  248. while (!wrapper.finished)
  249. STARPU_PTHREAD_COND_WAIT(&wrapper.cond, &wrapper.lock);
  250. STARPU_PTHREAD_MUTEX_UNLOCK(&wrapper.lock);
  251. }
  252. STARPU_PTHREAD_COND_DESTROY(&wrapper.cond);
  253. STARPU_PTHREAD_MUTEX_DESTROY(&wrapper.lock);
  254. /* At that moment, the caller holds a reference to the piece of data.
  255. * We enqueue the "post" sync task in the list associated to the handle
  256. * so that it is submitted by the starpu_data_release
  257. * function. */
  258. _starpu_add_post_sync_tasks(wrapper.post_sync_task, handle);
  259. _STARPU_LOG_OUT();
  260. return 0;
  261. }
  262. int starpu_data_acquire(starpu_data_handle_t handle, enum starpu_data_access_mode mode)
  263. {
  264. return starpu_data_acquire_on_node(handle, 0, mode);
  265. }
  266. /* This function must be called after starpu_data_acquire so that the
  267. * application release the data */
  268. void starpu_data_release_on_node(starpu_data_handle_t handle, unsigned node)
  269. {
  270. STARPU_ASSERT(handle);
  271. /* In case there are some implicit dependencies, unlock the "post sync" tasks */
  272. _starpu_unlock_post_sync_tasks(handle);
  273. /* The application can now release the rw-lock */
  274. _starpu_release_data_on_node(handle, 0, &handle->per_node[node]);
  275. }
  276. void starpu_data_release(starpu_data_handle_t handle)
  277. {
  278. starpu_data_release_on_node(handle, 0);
  279. }
  280. static void _prefetch_data_on_node(void *arg)
  281. {
  282. struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) arg;
  283. starpu_data_handle_t handle = wrapper->handle;
  284. int ret;
  285. struct _starpu_data_replicate *replicate = &handle->per_node[wrapper->node];
  286. ret = _starpu_fetch_data_on_node(handle, replicate, STARPU_R, wrapper->async, wrapper->async, NULL, NULL);
  287. STARPU_ASSERT(!ret);
  288. if (wrapper->async)
  289. free(wrapper);
  290. else
  291. {
  292. STARPU_PTHREAD_MUTEX_LOCK(&wrapper->lock);
  293. wrapper->finished = 1;
  294. STARPU_PTHREAD_COND_SIGNAL(&wrapper->cond);
  295. STARPU_PTHREAD_MUTEX_UNLOCK(&wrapper->lock);
  296. }
  297. _starpu_spin_lock(&handle->header_lock);
  298. if (!_starpu_notify_data_dependencies(handle))
  299. _starpu_spin_unlock(&handle->header_lock);
  300. }
  301. static
  302. int _starpu_prefetch_data_on_node_with_mode(starpu_data_handle_t handle, unsigned node, unsigned async, enum starpu_data_access_mode mode)
  303. {
  304. STARPU_ASSERT(handle);
  305. /* it is forbidden to call this function from a callback or a codelet */
  306. if (STARPU_UNLIKELY(!async && !_starpu_worker_may_perform_blocking_calls()))
  307. return -EDEADLK;
  308. struct user_interaction_wrapper *wrapper = (struct user_interaction_wrapper *) malloc(sizeof(*wrapper));
  309. wrapper->handle = handle;
  310. wrapper->node = node;
  311. wrapper->async = async;
  312. STARPU_PTHREAD_COND_INIT(&wrapper->cond, NULL);
  313. STARPU_PTHREAD_MUTEX_INIT(&wrapper->lock, NULL);
  314. wrapper->finished = 0;
  315. if (!_starpu_attempt_to_submit_data_request_from_apps(handle, mode, _prefetch_data_on_node, wrapper))
  316. {
  317. /* we can immediately proceed */
  318. struct _starpu_data_replicate *replicate = &handle->per_node[node];
  319. STARPU_PTHREAD_COND_DESTROY(&wrapper->cond);
  320. STARPU_PTHREAD_MUTEX_DESTROY(&wrapper->lock);
  321. free(wrapper);
  322. _starpu_fetch_data_on_node(handle, replicate, mode, async, async, NULL, NULL);
  323. /* remove the "lock"/reference */
  324. _starpu_spin_lock(&handle->header_lock);
  325. if (!async)
  326. {
  327. /* Release our refcnt, like _starpu_release_data_on_node would do */
  328. replicate->refcnt--;
  329. STARPU_ASSERT(replicate->refcnt >= 0);
  330. STARPU_ASSERT(handle->busy_count > 0);
  331. handle->busy_count--;
  332. }
  333. /* In case there was a temporary handle (eg. used for reduction), this
  334. * handle may have requested to be destroyed when the data is released
  335. * */
  336. if (!_starpu_notify_data_dependencies(handle))
  337. _starpu_spin_unlock(&handle->header_lock);
  338. }
  339. else if (!async)
  340. {
  341. STARPU_PTHREAD_MUTEX_LOCK(&wrapper->lock);
  342. while (!wrapper->finished)
  343. STARPU_PTHREAD_COND_WAIT(&wrapper->cond, &wrapper->lock);
  344. STARPU_PTHREAD_MUTEX_UNLOCK(&wrapper->lock);
  345. STARPU_PTHREAD_COND_DESTROY(&wrapper->cond);
  346. STARPU_PTHREAD_MUTEX_DESTROY(&wrapper->lock);
  347. free(wrapper);
  348. }
  349. return 0;
  350. }
  351. int starpu_data_prefetch_on_node(starpu_data_handle_t handle, unsigned node, unsigned async)
  352. {
  353. return _starpu_prefetch_data_on_node_with_mode(handle, node, async, STARPU_R);
  354. }
  355. /*
  356. * It is possible to specify that a piece of data can be discarded without
  357. * impacting the application.
  358. */
  359. void starpu_data_advise_as_important(starpu_data_handle_t handle, unsigned is_important)
  360. {
  361. _starpu_spin_lock(&handle->header_lock);
  362. /* first take all the children lock (in order !) */
  363. unsigned child;
  364. for (child = 0; child < handle->nchildren; child++)
  365. {
  366. /* make sure the intermediate children is advised as well */
  367. starpu_data_handle_t child_handle = starpu_data_get_child(handle, child);
  368. if (child_handle->nchildren > 0)
  369. starpu_data_advise_as_important(child_handle, is_important);
  370. }
  371. handle->is_not_important = !is_important;
  372. /* now the parent may be used again so we release the lock */
  373. _starpu_spin_unlock(&handle->header_lock);
  374. }
  375. void starpu_data_set_sequential_consistency_flag(starpu_data_handle_t handle, unsigned flag)
  376. {
  377. _starpu_spin_lock(&handle->header_lock);
  378. unsigned child;
  379. for (child = 0; child < handle->nchildren; child++)
  380. {
  381. /* make sure that the flags are applied to the children as well */
  382. starpu_data_handle_t child_handle = starpu_data_get_child(handle, child);
  383. if (child_handle->nchildren > 0)
  384. starpu_data_set_sequential_consistency_flag(child_handle, flag);
  385. }
  386. STARPU_PTHREAD_MUTEX_LOCK(&handle->sequential_consistency_mutex);
  387. handle->sequential_consistency = flag;
  388. STARPU_PTHREAD_MUTEX_UNLOCK(&handle->sequential_consistency_mutex);
  389. _starpu_spin_unlock(&handle->header_lock);
  390. }
  391. unsigned starpu_data_get_sequential_consistency_flag(starpu_data_handle_t handle)
  392. {
  393. return handle->sequential_consistency;
  394. }
  395. /* By default, sequential consistency is enabled */
  396. static unsigned default_sequential_consistency_flag = 1;
  397. unsigned starpu_data_get_default_sequential_consistency_flag(void)
  398. {
  399. return default_sequential_consistency_flag;
  400. }
  401. void starpu_data_set_default_sequential_consistency_flag(unsigned flag)
  402. {
  403. default_sequential_consistency_flag = flag;
  404. }
  405. /* Query the status of the handle on the specified memory node. */
  406. void starpu_data_query_status(starpu_data_handle_t handle, int memory_node, int *is_allocated, int *is_valid, int *is_requested)
  407. {
  408. #ifdef STARPU_DEVEL
  409. #warning FIXME
  410. #endif
  411. // _starpu_spin_lock(&handle->header_lock);
  412. if (is_allocated)
  413. *is_allocated = handle->per_node[memory_node].allocated;
  414. if (is_valid)
  415. *is_valid = (handle->per_node[memory_node].state != STARPU_INVALID);
  416. if (is_requested)
  417. {
  418. int requested = 0;
  419. unsigned node;
  420. for (node = 0; node < STARPU_MAXNODES; node++)
  421. {
  422. if (handle->per_node[memory_node].requested[node])
  423. {
  424. requested = 1;
  425. break;
  426. }
  427. }
  428. *is_requested = requested;
  429. }
  430. // _starpu_spin_unlock(&handle->header_lock);
  431. }