data_request.c 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745
  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 <starpu.h>
  18. #include <common/config.h>
  19. #include <common/utils.h>
  20. #include <datawizard/datawizard.h>
  21. /* TODO: This should be tuned according to driver capabilities
  22. * Data interfaces should also have to declare how many asynchronous requests
  23. * they have actually started (think of e.g. csr).
  24. */
  25. #define MAX_PENDING_REQUESTS_PER_NODE 20
  26. #define MAX_PENDING_PREFETCH_REQUESTS_PER_NODE 10
  27. /* requests that have not been treated at all */
  28. static struct _starpu_data_request_list *data_requests[STARPU_MAXNODES];
  29. static struct _starpu_data_request_list *prefetch_requests[STARPU_MAXNODES];
  30. static starpu_pthread_mutex_t data_requests_list_mutex[STARPU_MAXNODES];
  31. /* requests that are not terminated (eg. async transfers) */
  32. static struct _starpu_data_request_list *data_requests_pending[STARPU_MAXNODES];
  33. static unsigned data_requests_npending[STARPU_MAXNODES];
  34. static starpu_pthread_mutex_t data_requests_pending_list_mutex[STARPU_MAXNODES];
  35. void _starpu_init_data_request_lists(void)
  36. {
  37. unsigned i;
  38. for (i = 0; i < STARPU_MAXNODES; i++)
  39. {
  40. prefetch_requests[i] = _starpu_data_request_list_new();
  41. data_requests[i] = _starpu_data_request_list_new();
  42. /* Tell helgrind that we are fine with checking for list_empty
  43. * in _starpu_handle_node_data_requests, we will call it
  44. * periodically anyway */
  45. STARPU_HG_DISABLE_CHECKING(data_requests[i]);
  46. STARPU_HG_DISABLE_CHECKING(data_requests[i]->_head);
  47. STARPU_PTHREAD_MUTEX_INIT(&data_requests_list_mutex[i], NULL);
  48. data_requests_pending[i] = _starpu_data_request_list_new();
  49. data_requests_npending[i] = 0;
  50. STARPU_PTHREAD_MUTEX_INIT(&data_requests_pending_list_mutex[i], NULL);
  51. }
  52. STARPU_HG_DISABLE_CHECKING(data_requests_npending);
  53. }
  54. void _starpu_deinit_data_request_lists(void)
  55. {
  56. unsigned i;
  57. for (i = 0; i < STARPU_MAXNODES; i++)
  58. {
  59. STARPU_PTHREAD_MUTEX_DESTROY(&data_requests_pending_list_mutex[i]);
  60. _starpu_data_request_list_delete(data_requests_pending[i]);
  61. STARPU_PTHREAD_MUTEX_DESTROY(&data_requests_list_mutex[i]);
  62. _starpu_data_request_list_delete(data_requests[i]);
  63. _starpu_data_request_list_delete(prefetch_requests[i]);
  64. }
  65. }
  66. /* this should be called with the lock r->handle->header_lock taken */
  67. static void starpu_data_request_destroy(struct _starpu_data_request *r)
  68. {
  69. unsigned node;
  70. /* If this is a write only request, then there is no source and we use
  71. * the destination node to cache the request. Otherwise we store the
  72. * pending requests between src and dst. */
  73. if (r->mode & STARPU_R)
  74. {
  75. node = r->src_replicate->memory_node;
  76. }
  77. else
  78. {
  79. node = r->dst_replicate->memory_node;
  80. }
  81. STARPU_ASSERT(r->dst_replicate->request[node] == r);
  82. r->dst_replicate->request[node] = NULL;
  83. //fprintf(stderr, "DESTROY REQ %p (%d) refcnt %d\n", r, node, r->refcnt);
  84. _starpu_data_request_delete(r);
  85. }
  86. /* handle->lock should already be taken ! */
  87. struct _starpu_data_request *_starpu_create_data_request(starpu_data_handle_t handle,
  88. struct _starpu_data_replicate *src_replicate,
  89. struct _starpu_data_replicate *dst_replicate,
  90. unsigned handling_node,
  91. enum starpu_data_access_mode mode,
  92. unsigned ndeps,
  93. unsigned is_prefetch)
  94. {
  95. struct _starpu_data_request *r = _starpu_data_request_new();
  96. _starpu_spin_checklocked(&handle->header_lock);
  97. _starpu_spin_init(&r->lock);
  98. r->handle = handle;
  99. r->src_replicate = src_replicate;
  100. r->dst_replicate = dst_replicate;
  101. r->mode = mode;
  102. r->handling_node = handling_node;
  103. r->completed = 0;
  104. r->prefetch = is_prefetch;
  105. r->retval = -1;
  106. r->ndeps = ndeps;
  107. r->next_req_count = 0;
  108. r->callbacks = NULL;
  109. _starpu_spin_lock(&r->lock);
  110. /* Take a reference on the target for the request to be able to write it */
  111. dst_replicate->refcnt++;
  112. handle->busy_count++;
  113. if (mode & STARPU_R)
  114. {
  115. unsigned src_node = src_replicate->memory_node;
  116. dst_replicate->request[src_node] = r;
  117. /* Take a reference on the source for the request to be able to read it */
  118. src_replicate->refcnt++;
  119. handle->busy_count++;
  120. }
  121. else
  122. {
  123. unsigned dst_node = dst_replicate->memory_node;
  124. dst_replicate->request[dst_node] = r;
  125. }
  126. r->refcnt = 1;
  127. _starpu_spin_unlock(&r->lock);
  128. return r;
  129. }
  130. int _starpu_wait_data_request_completion(struct _starpu_data_request *r, unsigned may_alloc)
  131. {
  132. int retval;
  133. int do_delete = 0;
  134. unsigned local_node = _starpu_memory_node_get_local_key();
  135. do
  136. {
  137. _starpu_spin_lock(&r->lock);
  138. if (r->completed)
  139. break;
  140. _starpu_spin_unlock(&r->lock);
  141. #ifndef STARPU_NON_BLOCKING_DRIVERS
  142. _starpu_wake_all_blocked_workers_on_node(r->handling_node);
  143. #endif
  144. _starpu_datawizard_progress(local_node, may_alloc);
  145. }
  146. while (1);
  147. retval = r->retval;
  148. if (retval)
  149. _STARPU_DISP("REQUEST %p completed with retval %d!\n", r, r->retval);
  150. r->refcnt--;
  151. /* if nobody is waiting on that request, we can get rid of it */
  152. if (r->refcnt == 0)
  153. do_delete = 1;
  154. _starpu_spin_unlock(&r->lock);
  155. if (do_delete)
  156. starpu_data_request_destroy(r);
  157. return retval;
  158. }
  159. /* this is non blocking */
  160. void _starpu_post_data_request(struct _starpu_data_request *r, unsigned handling_node)
  161. {
  162. /* We don't have a worker for disk nodes, these should have been posted to a main RAM node */
  163. STARPU_ASSERT(starpu_node_get_kind(handling_node) != STARPU_DISK_RAM);
  164. STARPU_ASSERT(_starpu_memory_node_get_nworkers(handling_node));
  165. // _STARPU_DEBUG("POST REQUEST\n");
  166. /* If some dependencies are not fulfilled yet, we don't actually post the request */
  167. if (r->ndeps > 0)
  168. return;
  169. if (r->mode & STARPU_R)
  170. {
  171. STARPU_ASSERT(r->src_replicate->allocated);
  172. STARPU_ASSERT(r->src_replicate->refcnt);
  173. }
  174. /* insert the request in the proper list */
  175. STARPU_PTHREAD_MUTEX_LOCK(&data_requests_list_mutex[handling_node]);
  176. if (r->prefetch)
  177. _starpu_data_request_list_push_back(prefetch_requests[handling_node], r);
  178. else
  179. _starpu_data_request_list_push_back(data_requests[handling_node], r);
  180. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_list_mutex[handling_node]);
  181. #ifndef STARPU_NON_BLOCKING_DRIVERS
  182. _starpu_wake_all_blocked_workers_on_node(handling_node);
  183. #endif
  184. }
  185. /* We assume that r->lock is taken by the caller */
  186. void _starpu_data_request_append_callback(struct _starpu_data_request *r, void (*callback_func)(void *), void *callback_arg)
  187. {
  188. STARPU_ASSERT(r);
  189. if (callback_func)
  190. {
  191. struct _starpu_callback_list *link = (struct _starpu_callback_list *) malloc(sizeof(struct _starpu_callback_list));
  192. STARPU_ASSERT(link);
  193. link->callback_func = callback_func;
  194. link->callback_arg = callback_arg;
  195. link->next = r->callbacks;
  196. r->callbacks = link;
  197. }
  198. }
  199. /* This method is called with handle's header_lock taken, and unlocks it */
  200. static void starpu_handle_data_request_completion(struct _starpu_data_request *r)
  201. {
  202. unsigned do_delete = 0;
  203. starpu_data_handle_t handle = r->handle;
  204. enum starpu_data_access_mode mode = r->mode;
  205. struct _starpu_data_replicate *src_replicate = r->src_replicate;
  206. struct _starpu_data_replicate *dst_replicate = r->dst_replicate;
  207. #ifdef STARPU_MEMORY_STATS
  208. enum _starpu_cache_state old_src_replicate_state = src_replicate->state;
  209. #endif
  210. _starpu_spin_checklocked(&handle->header_lock);
  211. _starpu_update_data_state(handle, r->dst_replicate, mode);
  212. #ifdef STARPU_MEMORY_STATS
  213. if (src_replicate->state == STARPU_INVALID)
  214. {
  215. if (old_src_replicate_state == STARPU_OWNER)
  216. _starpu_memory_handle_stats_invalidated(handle, src_replicate->memory_node);
  217. else
  218. {
  219. /* XXX Currently only ex-OWNER are tagged as invalidated */
  220. /* XXX Have to check all old state of every node in case a SHARED data become OWNED by the dst_replicate */
  221. }
  222. }
  223. if (dst_replicate->state == STARPU_SHARED)
  224. _starpu_memory_handle_stats_loaded_shared(handle, dst_replicate->memory_node);
  225. else if (dst_replicate->state == STARPU_OWNER)
  226. {
  227. _starpu_memory_handle_stats_loaded_owner(handle, dst_replicate->memory_node);
  228. }
  229. #endif
  230. #ifdef STARPU_USE_FXT
  231. unsigned src_node = src_replicate->memory_node;
  232. unsigned dst_node = dst_replicate->memory_node;
  233. size_t size = _starpu_data_get_size(handle);
  234. _STARPU_TRACE_END_DRIVER_COPY(src_node, dst_node, size, r->com_id);
  235. #endif
  236. /* Once the request has been fulfilled, we may submit the requests that
  237. * were chained to that request. */
  238. unsigned chained_req;
  239. for (chained_req = 0; chained_req < r->next_req_count; chained_req++)
  240. {
  241. struct _starpu_data_request *next_req = r->next_req[chained_req];
  242. STARPU_ASSERT(next_req->ndeps > 0);
  243. next_req->ndeps--;
  244. _starpu_post_data_request(next_req, next_req->handling_node);
  245. }
  246. r->completed = 1;
  247. /* Remove a reference on the destination replicate for the request */
  248. STARPU_ASSERT(dst_replicate->refcnt > 0);
  249. dst_replicate->refcnt--;
  250. STARPU_ASSERT(handle->busy_count > 0);
  251. handle->busy_count--;
  252. /* In case the source was "locked" by the request too */
  253. if (mode & STARPU_R)
  254. {
  255. STARPU_ASSERT(src_replicate->refcnt > 0);
  256. src_replicate->refcnt--;
  257. STARPU_ASSERT(handle->busy_count > 0);
  258. handle->busy_count--;
  259. }
  260. unsigned destroyed = _starpu_data_check_not_busy(handle);
  261. r->refcnt--;
  262. /* if nobody is waiting on that request, we can get rid of it */
  263. if (r->refcnt == 0)
  264. do_delete = 1;
  265. r->retval = 0;
  266. /* In case there are one or multiple callbacks, we execute them now. */
  267. struct _starpu_callback_list *callbacks = r->callbacks;
  268. _starpu_spin_unlock(&r->lock);
  269. if (do_delete)
  270. starpu_data_request_destroy(r);
  271. if (!destroyed)
  272. _starpu_spin_unlock(&handle->header_lock);
  273. /* We do the callback once the lock is released so that they can do
  274. * blocking operations with the handle (eg. release it) */
  275. while (callbacks)
  276. {
  277. callbacks->callback_func(callbacks->callback_arg);
  278. struct _starpu_callback_list *next = callbacks->next;
  279. free(callbacks);
  280. callbacks = next;
  281. }
  282. }
  283. /* TODO : accounting to see how much time was spent working for other people ... */
  284. static int starpu_handle_data_request(struct _starpu_data_request *r, unsigned may_alloc, int prefetch)
  285. {
  286. starpu_data_handle_t handle = r->handle;
  287. if (_starpu_spin_trylock(&handle->header_lock))
  288. return -EBUSY;
  289. if (_starpu_spin_trylock(&r->lock))
  290. {
  291. _starpu_spin_unlock(&handle->header_lock);
  292. return -EBUSY;
  293. }
  294. struct _starpu_data_replicate *src_replicate = r->src_replicate;
  295. struct _starpu_data_replicate *dst_replicate = r->dst_replicate;
  296. enum starpu_data_access_mode r_mode = r->mode;
  297. STARPU_ASSERT(!(r_mode & STARPU_R) || src_replicate);
  298. STARPU_ASSERT(!(r_mode & STARPU_R) || src_replicate->allocated);
  299. STARPU_ASSERT(!(r_mode & STARPU_R) || src_replicate->refcnt);
  300. _starpu_spin_unlock(&r->lock);
  301. /* FIXME: the request may get upgraded from here to freeing it... */
  302. /* perform the transfer */
  303. /* the header of the data must be locked by the worker that submitted the request */
  304. r->retval = _starpu_driver_copy_data_1_to_1(handle, src_replicate,
  305. dst_replicate, !(r_mode & STARPU_R), r, may_alloc);
  306. if (r->retval == -ENOMEM)
  307. {
  308. /* If there was not enough memory, we will try to redo the
  309. * request later. */
  310. _starpu_spin_unlock(&handle->header_lock);
  311. return -ENOMEM;
  312. }
  313. if (r->retval == -EAGAIN)
  314. {
  315. /* The request was successful, but could not be terminted
  316. * immediatly. We will handle the completion of the request
  317. * asynchronously. The request is put in the list of "pending"
  318. * requests in the meantime. */
  319. _starpu_spin_unlock(&handle->header_lock);
  320. STARPU_PTHREAD_MUTEX_LOCK(&data_requests_pending_list_mutex[r->handling_node]);
  321. _starpu_data_request_list_push_front(data_requests_pending[r->handling_node], r);
  322. data_requests_npending[r->handling_node]++;
  323. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_pending_list_mutex[r->handling_node]);
  324. return -EAGAIN;
  325. }
  326. /* the request has been handled */
  327. _starpu_spin_lock(&r->lock);
  328. starpu_handle_data_request_completion(r);
  329. return 0;
  330. }
  331. int _starpu_handle_node_data_requests(unsigned src_node, unsigned may_alloc)
  332. {
  333. struct _starpu_data_request *r;
  334. struct _starpu_data_request_list *new_data_requests;
  335. struct _starpu_data_request_list *empty_list;
  336. int ret = 0;
  337. /* Here helgrind would should that this is an un protected access.
  338. * We however don't care about missing an entry, we will get called
  339. * again sooner or later. */
  340. if (_starpu_data_request_list_empty(data_requests[src_node]))
  341. return 0;
  342. empty_list = _starpu_data_request_list_new();
  343. /* take all the entries from the request list */
  344. if (STARPU_PTHREAD_MUTEX_TRYLOCK(&data_requests_list_mutex[src_node]))
  345. /* List is busy, do not bother with it */
  346. return -EBUSY;
  347. struct _starpu_data_request_list *local_list = data_requests[src_node];
  348. if (_starpu_data_request_list_empty(local_list))
  349. {
  350. /* there is no request */
  351. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_list_mutex[src_node]);
  352. _starpu_data_request_list_delete(empty_list);
  353. return 0;
  354. }
  355. /* There is an entry: we create a new empty list to replace the list of
  356. * requests, and we handle the request(s) one by one in the former
  357. * list, without concurrency issues.*/
  358. data_requests[src_node] = empty_list;
  359. STARPU_HG_DISABLE_CHECKING(data_requests[src_node]->_head);
  360. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_list_mutex[src_node]);
  361. new_data_requests = _starpu_data_request_list_new();
  362. /* for all entries of the list */
  363. while (!_starpu_data_request_list_empty(local_list))
  364. {
  365. int res;
  366. if (data_requests_npending[src_node] >= MAX_PENDING_REQUESTS_PER_NODE)
  367. {
  368. /* Too many requests at the same time, skip pushing
  369. * more for now */
  370. ret = -EBUSY;
  371. break;
  372. }
  373. r = _starpu_data_request_list_pop_front(local_list);
  374. res = starpu_handle_data_request(r, may_alloc, 0);
  375. if (res != 0 && res != -EAGAIN)
  376. {
  377. /* handle is busy, or not enough memory, postpone for now */
  378. ret = res;
  379. _starpu_data_request_list_push_back(new_data_requests, r);
  380. break;
  381. }
  382. }
  383. while (!_starpu_data_request_list_empty(local_list))
  384. {
  385. r = _starpu_data_request_list_pop_front(local_list);
  386. _starpu_data_request_list_push_back(new_data_requests, r);
  387. }
  388. if (!_starpu_data_request_list_empty(new_data_requests))
  389. {
  390. STARPU_PTHREAD_MUTEX_LOCK(&data_requests_list_mutex[src_node]);
  391. _starpu_data_request_list_push_list_front(new_data_requests, data_requests[src_node]);
  392. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_list_mutex[src_node]);
  393. }
  394. _starpu_data_request_list_delete(new_data_requests);
  395. _starpu_data_request_list_delete(local_list);
  396. return ret;
  397. }
  398. void _starpu_handle_node_prefetch_requests(unsigned src_node, unsigned may_alloc)
  399. {
  400. struct _starpu_data_request *r;
  401. struct _starpu_data_request_list *new_data_requests;
  402. struct _starpu_data_request_list *new_prefetch_requests;
  403. struct _starpu_data_request_list *empty_list;
  404. if (_starpu_data_request_list_empty(prefetch_requests[src_node]))
  405. return;
  406. empty_list = _starpu_data_request_list_new();
  407. /* take all the entries from the request list */
  408. if (STARPU_PTHREAD_MUTEX_TRYLOCK(&data_requests_list_mutex[src_node]))
  409. /* List is busy, do not bother with it */
  410. return;
  411. struct _starpu_data_request_list *local_list = prefetch_requests[src_node];
  412. if (_starpu_data_request_list_empty(local_list))
  413. {
  414. /* there is no request */
  415. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_list_mutex[src_node]);
  416. _starpu_data_request_list_delete(empty_list);
  417. return;
  418. }
  419. /* There is an entry: we create a new empty list to replace the list of
  420. * requests, and we handle the request(s) one by one in the former
  421. * list, without concurrency issues.*/
  422. prefetch_requests[src_node] = empty_list;
  423. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_list_mutex[src_node]);
  424. new_data_requests = _starpu_data_request_list_new();
  425. new_prefetch_requests = _starpu_data_request_list_new();
  426. /* for all entries of the list */
  427. while (!_starpu_data_request_list_empty(local_list))
  428. {
  429. int res;
  430. if (data_requests_npending[src_node] >= MAX_PENDING_PREFETCH_REQUESTS_PER_NODE)
  431. {
  432. /* Too many requests at the same time, skip pushing
  433. * more for now */
  434. break;
  435. }
  436. r = _starpu_data_request_list_pop_front(local_list);
  437. res = starpu_handle_data_request(r, may_alloc, 1);
  438. if (res != 0 && res != -EAGAIN)
  439. {
  440. if (r->prefetch)
  441. _starpu_data_request_list_push_back(new_prefetch_requests, r);
  442. else
  443. {
  444. /* Prefetch request promoted while in tmp list*/
  445. _starpu_data_request_list_push_back(new_data_requests, r);
  446. }
  447. break;
  448. }
  449. }
  450. while(!_starpu_data_request_list_empty(local_list))
  451. {
  452. r = _starpu_data_request_list_pop_front(local_list);
  453. if (r->prefetch)
  454. _starpu_data_request_list_push_back(new_prefetch_requests, r);
  455. else
  456. /* Prefetch request promoted while in tmp list*/
  457. _starpu_data_request_list_push_back(new_data_requests, r);
  458. }
  459. if (!(_starpu_data_request_list_empty(new_data_requests) && _starpu_data_request_list_empty(new_prefetch_requests)))
  460. {
  461. STARPU_PTHREAD_MUTEX_LOCK(&data_requests_list_mutex[src_node]);
  462. if (!(_starpu_data_request_list_empty(new_data_requests)))
  463. _starpu_data_request_list_push_list_front(new_data_requests, data_requests[src_node]);
  464. if (!(_starpu_data_request_list_empty(new_prefetch_requests)))
  465. _starpu_data_request_list_push_list_front(new_prefetch_requests, prefetch_requests[src_node]);
  466. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_list_mutex[src_node]);
  467. }
  468. _starpu_data_request_list_delete(new_data_requests);
  469. _starpu_data_request_list_delete(new_prefetch_requests);
  470. _starpu_data_request_list_delete(local_list);
  471. }
  472. static void _handle_pending_node_data_requests(unsigned src_node, unsigned force)
  473. {
  474. // _STARPU_DEBUG("_starpu_handle_pending_node_data_requests ...\n");
  475. //
  476. struct _starpu_data_request_list *new_data_requests_pending;
  477. struct _starpu_data_request_list *empty_list;
  478. unsigned taken, kept;
  479. if (_starpu_data_request_list_empty(data_requests_pending[src_node]))
  480. return;
  481. empty_list = _starpu_data_request_list_new();
  482. if (STARPU_PTHREAD_MUTEX_TRYLOCK(&data_requests_pending_list_mutex[src_node]) && !force)
  483. /* List is busy, do not bother with it */
  484. return;
  485. /* for all entries of the list */
  486. struct _starpu_data_request_list *local_list = data_requests_pending[src_node];
  487. if (_starpu_data_request_list_empty(local_list))
  488. {
  489. /* there is no request */
  490. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_pending_list_mutex[src_node]);
  491. _starpu_data_request_list_delete(empty_list);
  492. return;
  493. }
  494. data_requests_pending[src_node] = empty_list;
  495. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_pending_list_mutex[src_node]);
  496. new_data_requests_pending = _starpu_data_request_list_new();
  497. taken = 0;
  498. kept = 0;
  499. while (!_starpu_data_request_list_empty(local_list))
  500. {
  501. struct _starpu_data_request *r;
  502. r = _starpu_data_request_list_pop_front(local_list);
  503. taken++;
  504. starpu_data_handle_t handle = r->handle;
  505. if (force)
  506. /* Have to wait for the handle, whatever it takes */
  507. _starpu_spin_lock(&handle->header_lock);
  508. else
  509. if (_starpu_spin_trylock(&handle->header_lock))
  510. {
  511. /* Handle is busy, retry this later */
  512. _starpu_data_request_list_push_back(new_data_requests_pending, r);
  513. kept++;
  514. continue;
  515. }
  516. /* This shouldn't be too hard to acquire */
  517. _starpu_spin_lock(&r->lock);
  518. /* wait until the transfer is terminated */
  519. if (force)
  520. {
  521. _starpu_driver_wait_request_completion(&r->async_channel);
  522. starpu_handle_data_request_completion(r);
  523. }
  524. else
  525. {
  526. if (_starpu_driver_test_request_completion(&r->async_channel))
  527. {
  528. /* The request was completed */
  529. starpu_handle_data_request_completion(r);
  530. }
  531. else
  532. {
  533. /* The request was not completed, so we put it
  534. * back again on the list of pending requests
  535. * so that it can be handled later on. */
  536. _starpu_spin_unlock(&r->lock);
  537. _starpu_spin_unlock(&handle->header_lock);
  538. _starpu_data_request_list_push_back(new_data_requests_pending, r);
  539. kept++;
  540. }
  541. }
  542. }
  543. STARPU_PTHREAD_MUTEX_LOCK(&data_requests_pending_list_mutex[src_node]);
  544. data_requests_npending[src_node] -= taken - kept;
  545. if (kept)
  546. _starpu_data_request_list_push_list_back(data_requests_pending[src_node], new_data_requests_pending);
  547. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_pending_list_mutex[src_node]);
  548. _starpu_data_request_list_delete(local_list);
  549. _starpu_data_request_list_delete(new_data_requests_pending);
  550. }
  551. void _starpu_handle_pending_node_data_requests(unsigned src_node)
  552. {
  553. _handle_pending_node_data_requests(src_node, 0);
  554. }
  555. void _starpu_handle_all_pending_node_data_requests(unsigned src_node)
  556. {
  557. _handle_pending_node_data_requests(src_node, 1);
  558. }
  559. int _starpu_check_that_no_data_request_exists(unsigned node)
  560. {
  561. /* XXX lock that !!! that's a quick'n'dirty test */
  562. int no_request;
  563. int no_pending;
  564. STARPU_PTHREAD_MUTEX_LOCK(&data_requests_list_mutex[node]);
  565. no_request = _starpu_data_request_list_empty(data_requests[node]);
  566. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_list_mutex[node]);
  567. STARPU_PTHREAD_MUTEX_LOCK(&data_requests_pending_list_mutex[node]);
  568. no_pending = !data_requests_npending[node];
  569. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_pending_list_mutex[node]);
  570. return (no_request && no_pending);
  571. }
  572. void _starpu_update_prefetch_status(struct _starpu_data_request *r)
  573. {
  574. STARPU_ASSERT(r->prefetch > 0);
  575. r->prefetch=0;
  576. /* We have to promote chained_request too! */
  577. unsigned chained_req;
  578. for (chained_req = 0; chained_req < r->next_req_count; chained_req++)
  579. {
  580. struct _starpu_data_request *next_req = r->next_req[chained_req];
  581. if (next_req->prefetch)
  582. _starpu_update_prefetch_status(next_req);
  583. }
  584. STARPU_PTHREAD_MUTEX_LOCK(&data_requests_list_mutex[r->handling_node]);
  585. /* The request can be in a different list (handling request or the temp list)
  586. * we have to check that it is really in the prefetch list. */
  587. struct _starpu_data_request *r_iter;
  588. for (r_iter = _starpu_data_request_list_begin(prefetch_requests[r->handling_node]);
  589. r_iter != _starpu_data_request_list_end(prefetch_requests[r->handling_node]);
  590. r_iter = _starpu_data_request_list_next(r_iter))
  591. {
  592. if (r==r_iter)
  593. {
  594. _starpu_data_request_list_erase(prefetch_requests[r->handling_node],r);
  595. _starpu_data_request_list_push_front(data_requests[r->handling_node],r);
  596. break;
  597. }
  598. }
  599. STARPU_PTHREAD_MUTEX_UNLOCK(&data_requests_list_mutex[r->handling_node]);
  600. }