starpu_replay.c 30 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2016-2021 Université de Bordeaux, CNRS (LaBRI UMR 5800), Inria
  4. * Copyright (C) 2017 Erwan Leria
  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. /*
  18. * This reads a tasks.rec file and replays the recorded task graph.
  19. * Currently, this version is done to run with simgrid.
  20. *
  21. * For further information, contact erwan.leria@inria.fr
  22. */
  23. #include <starpu.h>
  24. #include <unistd.h>
  25. #include <stdio.h>
  26. #include <math.h>
  27. #include <common/uthash.h>
  28. #include <common/utils.h>
  29. #include <starpu_scheduler.h>
  30. #include <common/rbtree.h>
  31. #define REPLAY_NMAX_DEPENDENCIES 8
  32. #define ARRAY_DUP(in, out, n) memcpy(out, in, n * sizeof(*out))
  33. #define ARRAY_INIT(array, n) memset(array, 0, n * sizeof(*array))
  34. /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
  35. * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
  36. * Declarations of global variables, structures, pointers, ... *
  37. * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
  38. * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
  39. static int static_workerid;
  40. /* TODO: move to core header while moving starpu_replay_sched to core */
  41. extern void schedRecInit(const char * filename);
  42. extern void applySchedRec(struct starpu_task * starpu_task, long submit_order);
  43. /* Enum for normal and "wontuse" tasks */
  44. enum task_type {NormalTask, WontUseTask};
  45. typedef unsigned long jobid_t;
  46. enum task_type control;
  47. static char *name = NULL;
  48. static char *model = NULL;
  49. static jobid_t jobid;
  50. static jobid_t *dependson;
  51. static long submitorder = -1;
  52. static starpu_tag_t tag;
  53. static int workerid;
  54. static uint32_t footprint;
  55. static double flops, total_flops = 0.;
  56. static double startTime; //start time (The instant when the task starts)
  57. static double endTime; //end time (The instant when the task ends)
  58. static int iteration = -1;
  59. static starpu_data_handle_t handles[STARPU_NMAXBUFS];
  60. static enum starpu_data_access_mode modes[STARPU_NMAXBUFS];
  61. static char normal_reg_signal[STARPU_NMAXBUFS];
  62. /* Use the following arrays when the number of data is greater than STARPU_NMAXBUFS */
  63. starpu_data_handle_t * handles_ptr;
  64. enum starpu_data_access_mode * modes_ptr;
  65. size_t * sizes_set;
  66. static size_t dependson_size;
  67. static size_t ndependson;
  68. static unsigned nb_parameters = 0; /* Number of parameters */
  69. static int alloc_mode; /* If alloc_mode value is 1, then the handles are stored in dyn_handles, else they are in handles */
  70. static int priority = 0;
  71. char * reg_signal = NULL; /* The register signal (0 or 1 coded on 8 bit) is used to know which handle of the task has to be registered in StarPU (in fact to avoid handle twice)*/
  72. /* Record all tasks, hashed by jobid. */
  73. static struct task
  74. {
  75. struct starpu_rbtree_node node;
  76. UT_hash_handle hh;
  77. jobid_t jobid;
  78. int iteration;
  79. long submit_order;
  80. jobid_t *deps;
  81. size_t ndependson;
  82. struct starpu_task task;
  83. enum task_type type;
  84. int reg_signal;
  85. } *tasks;
  86. /* Record handles */
  87. static struct handle
  88. {
  89. UT_hash_handle hh;
  90. starpu_data_handle_t mem_ptr; /* This value should be the registered handle */
  91. starpu_data_handle_t handle; /* The key is the original value of the handle in the file */
  92. } * handles_hash;
  93. /* Record models */
  94. static struct perfmodel
  95. {
  96. UT_hash_handle hh;
  97. struct starpu_perfmodel perfmodel;
  98. char * model_name;
  99. } * model_hash;
  100. /*
  101. * Replay data interface
  102. * We don't care about many things anyway, essentially only sizes.
  103. */
  104. struct replay_interface
  105. {
  106. enum starpu_data_interface_id id;
  107. starpu_data_handle_t orig_handle;
  108. size_t size;
  109. size_t alloc_size;
  110. size_t max_size;
  111. };
  112. static struct starpu_data_interface_ops replay_interface_ops;
  113. static void register_replay(starpu_data_handle_t handle, unsigned home_node, void *data_interface)
  114. {
  115. (void) home_node;
  116. struct replay_interface *replay_interface = data_interface;
  117. unsigned node;
  118. for (node = 0; node < STARPU_MAXNODES; node++)
  119. {
  120. struct replay_interface *local_interface =
  121. starpu_data_get_interface_on_node(handle, node);
  122. local_interface->id = replay_interface->id;
  123. local_interface->orig_handle = replay_interface->orig_handle;
  124. local_interface->size = replay_interface->size;
  125. local_interface->alloc_size = replay_interface->alloc_size;
  126. local_interface->max_size = replay_interface->max_size;
  127. }
  128. }
  129. static void replay_data_register(starpu_data_handle_t *handleptr, starpu_data_handle_t orig_handle, int home_node, size_t size, size_t alloc_size, size_t max_size)
  130. {
  131. if (replay_interface_ops.interfaceid == STARPU_UNKNOWN_INTERFACE_ID)
  132. {
  133. replay_interface_ops.interfaceid = starpu_data_interface_get_next_id();
  134. }
  135. struct replay_interface interface =
  136. {
  137. .id = replay_interface_ops.interfaceid,
  138. .orig_handle = orig_handle,
  139. .size = size,
  140. .alloc_size = alloc_size,
  141. .max_size = max_size,
  142. };
  143. starpu_data_register(handleptr, home_node, &interface, &replay_interface_ops);
  144. }
  145. static size_t replay_get_size(starpu_data_handle_t handle)
  146. {
  147. struct replay_interface *interface =
  148. starpu_data_get_interface_on_node(handle, STARPU_MAIN_RAM);
  149. return interface->size;
  150. }
  151. static size_t replay_get_alloc_size(starpu_data_handle_t handle)
  152. {
  153. struct replay_interface *interface =
  154. starpu_data_get_interface_on_node(handle, STARPU_MAIN_RAM);
  155. return interface->alloc_size;
  156. }
  157. static size_t replay_get_max_size(starpu_data_handle_t handle)
  158. {
  159. struct replay_interface *interface =
  160. starpu_data_get_interface_on_node(handle, STARPU_MAIN_RAM);
  161. return interface->max_size;
  162. }
  163. static uint32_t replay_footprint(starpu_data_handle_t handle)
  164. {
  165. return starpu_hash_crc32c_be(replay_get_size(handle), 0);
  166. }
  167. static int replay_compare(void *data_interface_a, void *data_interface_b)
  168. {
  169. struct replay_interface *replay_a = data_interface_a;
  170. struct replay_interface *replay_b = data_interface_b;
  171. /* Two variables are considered compatible if they have the same size */
  172. return replay_a->size == replay_b->size;
  173. }
  174. static void display_replay(starpu_data_handle_t handle, FILE *f)
  175. {
  176. struct replay_interface *replay_interface =
  177. starpu_data_get_interface_on_node(handle, STARPU_MAIN_RAM);
  178. fprintf(f, "%lu/%lu/%lu\t",
  179. (unsigned long) replay_interface->size,
  180. (unsigned long) replay_interface->alloc_size,
  181. (unsigned long) replay_interface->max_size);
  182. }
  183. static starpu_ssize_t describe_replay(void *data_interface, char *buf, size_t size)
  184. {
  185. struct replay_interface *replay_interface = data_interface;
  186. return snprintf(buf, size, "r%lu/%lu/%lu\t",
  187. (unsigned long) replay_interface->size,
  188. (unsigned long) replay_interface->alloc_size,
  189. (unsigned long) replay_interface->max_size);
  190. }
  191. static starpu_ssize_t allocate_replay_on_node(void *data_interface, unsigned dst_node)
  192. {
  193. struct replay_interface *replay_interface = data_interface;
  194. starpu_memory_allocate(dst_node, replay_interface->alloc_size, STARPU_MEMORY_OVERFLOW);
  195. return 0;
  196. }
  197. static void free_replay_on_node(void *data_interface, unsigned dst_node)
  198. {
  199. struct replay_interface *replay_interface = data_interface;
  200. starpu_memory_deallocate(dst_node, replay_interface->alloc_size);
  201. }
  202. static int replay_copy(void *src_interface, unsigned src_node, void *dst_interface, unsigned dst_node, void *async_data)
  203. {
  204. (void) dst_interface;
  205. struct replay_interface *src = src_interface;
  206. /* We don't care about pointers */
  207. return starpu_interface_copy(1, 0, src_node, 1, 0, dst_node, src->size, async_data);
  208. }
  209. static const struct starpu_data_copy_methods replay_copy_data_methods =
  210. {
  211. .any_to_any = replay_copy,
  212. };
  213. static struct starpu_data_interface_ops replay_interface_ops =
  214. {
  215. .register_data_handle = register_replay,
  216. .allocate_data_on_node = allocate_replay_on_node,
  217. .free_data_on_node = free_replay_on_node,
  218. .copy_methods = &replay_copy_data_methods,
  219. .get_size = replay_get_size,
  220. .get_alloc_size = replay_get_alloc_size,
  221. .get_max_size = replay_get_max_size,
  222. .footprint = replay_footprint,
  223. .compare = replay_compare,
  224. .interfaceid = STARPU_UNKNOWN_INTERFACE_ID,
  225. .interface_size = sizeof(struct replay_interface),
  226. .display = display_replay,
  227. .pack_data = NULL,
  228. .peek_data = NULL,
  229. .unpack_data = NULL,
  230. .describe = describe_replay,
  231. /* We want to observe actual allocations/deallocations */
  232. .dontcache = 1,
  233. };
  234. /* [SUBMITORDER] The tree of the submit order */
  235. static struct starpu_rbtree tree = STARPU_RBTREE_INITIALIZER;
  236. /* the cmp_fn arg for rb_tree_insert() */
  237. unsigned int diff(struct starpu_rbtree_node * left_elm, struct starpu_rbtree_node * right_elm)
  238. {
  239. long oleft = ((struct task *) left_elm)->submit_order;
  240. long oright = ((struct task *) right_elm)->submit_order;
  241. if (oleft == -1 && oright == -1)
  242. {
  243. if (left_elm < right_elm)
  244. return -1;
  245. else
  246. return 1;
  247. }
  248. return oleft - oright;
  249. }
  250. /* Settings for the perfmodel */
  251. struct task_arg
  252. {
  253. uint32_t footprint;
  254. unsigned narch;
  255. double perf[];
  256. };
  257. uint32_t get_footprint(struct starpu_task * task)
  258. {
  259. return ((struct task_arg*) (task->cl_arg))->footprint;
  260. }
  261. double arch_cost_function(struct starpu_task *task, struct starpu_perfmodel_arch *arch, unsigned nimpl)
  262. {
  263. int device = starpu_perfmodel_arch_comb_get(arch->ndevices, arch->devices);
  264. STARPU_ASSERT(device != -1);
  265. (void) nimpl;
  266. /* Then, get the pointer to the value of the expected time */
  267. struct task_arg *arg = task->cl_arg;
  268. if (device < (int) arg->narch)
  269. {
  270. double val = arg->perf[device];
  271. if (!(val == 0 || isnan(val)))
  272. return val;
  273. }
  274. fprintf(stderr, "[starpu] Error, expected_time is 0 or lower (replay.c line : %d)", __LINE__- 6);
  275. return 0.0;
  276. }
  277. /* End of settings */
  278. static unsigned long nexecuted_tasks;
  279. void dumb_kernel(void *buffers[], void *args)
  280. {
  281. (void) buffers;
  282. (void) args;
  283. nexecuted_tasks++;
  284. if (!(nexecuted_tasks % 1000))
  285. {
  286. fprintf(stderr, "\rExecuted task %lu...", nexecuted_tasks);
  287. fflush(stdout);
  288. }
  289. unsigned this_worker = starpu_worker_get_id_check();
  290. struct starpu_perfmodel_arch *perf_arch = starpu_worker_get_perf_archtype(this_worker, STARPU_NMAX_SCHED_CTXS);
  291. struct starpu_task *task = starpu_task_get_current();
  292. unsigned impl = starpu_task_get_implementation(task);
  293. double length = starpu_task_expected_length(task, perf_arch, impl);
  294. STARPU_ASSERT_MSG(!_STARPU_IS_ZERO(length) && !isnan(length),
  295. "Codelet %s does not have a perfmodel, or is not calibrated enough, please re-run in non-simgrid mode until it is calibrated",
  296. starpu_task_get_name(task));
  297. starpu_sleep(length / 1000000);
  298. }
  299. /* [CODELET] Initialization of an unique codelet for all the tasks*/
  300. static int can_execute(unsigned worker_id, struct starpu_task *task, unsigned nimpl)
  301. {
  302. struct starpu_perfmodel_arch * arch = starpu_worker_get_perf_archtype(worker_id, STARPU_NMAX_SCHED_CTXS);
  303. int device = starpu_perfmodel_arch_comb_get(arch->ndevices, arch->devices);
  304. if (device == -1)
  305. /* Doesn't exist yet, thus unknown, assuming it can not work there. */
  306. return 0;
  307. (void) nimpl;
  308. /* Then, get the pointer to the value of the expected time */
  309. struct task_arg *arg = task->cl_arg;
  310. if (device < (int) arg->narch)
  311. {
  312. double val = arg->perf[device];
  313. if (!(val == 0 || isnan(val)))
  314. return 1;
  315. }
  316. return 0;
  317. }
  318. static struct starpu_perfmodel myperfmodel =
  319. {
  320. .type = STARPU_PER_ARCH,
  321. .arch_cost_function = arch_cost_function,
  322. .footprint = get_footprint,
  323. };
  324. static struct starpu_codelet cl =
  325. {
  326. .cpu_funcs = { dumb_kernel },
  327. .cpu_funcs_name = { "dumb_kernel" },
  328. .cuda_funcs = { dumb_kernel },
  329. .opencl_funcs = { dumb_kernel },
  330. .nbuffers = STARPU_VARIABLE_NBUFFERS,
  331. .can_execute = can_execute,
  332. .model = &myperfmodel,
  333. .flags = STARPU_CODELET_SIMGRID_EXECUTE,
  334. };
  335. /* * * * * * * * * * * * * *
  336. * * * * * Functions * * * * *
  337. * * * * * * * * * * * * * * */
  338. /* The following function checks if the program has to use static or dynamic arrays*/
  339. static int set_alloc_mode(int total_parameters)
  340. {
  341. return total_parameters <= STARPU_NMAXBUFS;
  342. }
  343. /* According to the allocation mode, modify handles_ptr and modes_ptr in static or dynamic */
  344. static void arrays_managing(int mode)
  345. {
  346. if (mode)
  347. {
  348. handles_ptr = &handles[0];
  349. modes_ptr = &modes[0];
  350. reg_signal = &normal_reg_signal[0];
  351. }
  352. else
  353. {
  354. _STARPU_MALLOC(handles_ptr, sizeof(*handles_ptr) * nb_parameters);
  355. _STARPU_MALLOC(modes_ptr, sizeof(*modes_ptr) * nb_parameters);
  356. _STARPU_CALLOC(reg_signal, nb_parameters, sizeof(char));
  357. }
  358. }
  359. static unsigned count_number_tokens(const char* buffer, const char* delim)
  360. {
  361. char* dup = strdup(buffer);
  362. int result = 0;
  363. char* token = strtok(dup, delim);
  364. while(token != NULL)
  365. {
  366. ++result;
  367. token = strtok(NULL, delim);
  368. }
  369. free(dup);
  370. return result;
  371. }
  372. /* Check if a handle hasn't been registered yet */
  373. static void variable_data_register_check(size_t * array_of_size, int nb_handles)
  374. {
  375. int h, i;
  376. starpu_data_handle_t orig_handles[nb_handles];
  377. ARRAY_DUP(handles_ptr, orig_handles, nb_handles);
  378. for (h = 0 ; h < nb_handles ; h++)
  379. {
  380. if(reg_signal[h]) /* Get the register signal, if it's 1 do ... */
  381. {
  382. struct handle * handles_cell;
  383. for (i = 0; i < h; i++)
  384. {
  385. /* Maybe we just registered it in this very h loop */
  386. if (handles_ptr[h] == orig_handles[i])
  387. {
  388. handles_ptr[h] = handles_ptr[i];
  389. break;
  390. }
  391. }
  392. if (i == h)
  393. {
  394. _STARPU_MALLOC(handles_cell, sizeof(*handles_cell));
  395. STARPU_ASSERT(handles_cell != NULL);
  396. handles_cell->handle = handles_ptr[h]; /* Get the hidden key (initial handle from the file) to store it as a key*/
  397. replay_data_register(handles_ptr+h, handles_ptr[h],
  398. modes_ptr[h] & STARPU_R ? STARPU_MAIN_RAM : -1,
  399. array_of_size[h], array_of_size[h], array_of_size[h]);
  400. handles_cell->mem_ptr = handles_ptr[h]; /* Store the new value of the handle into the hash table */
  401. HASH_ADD(hh, handles_hash, handle, sizeof(handles_ptr[h]), handles_cell);
  402. }
  403. }
  404. }
  405. }
  406. void reset(void)
  407. {
  408. control = NormalTask;
  409. if (name != NULL)
  410. {
  411. free(name);
  412. name = NULL;
  413. }
  414. if (model != NULL)
  415. {
  416. free(model);
  417. model = NULL;
  418. }
  419. if (sizes_set != NULL)
  420. {
  421. free(sizes_set);
  422. sizes_set = NULL;
  423. }
  424. if (reg_signal != NULL)
  425. {
  426. if (!alloc_mode)
  427. {
  428. free(reg_signal);
  429. reg_signal = NULL;
  430. }
  431. else
  432. {
  433. ARRAY_INIT(reg_signal, nb_parameters);
  434. }
  435. }
  436. jobid = 0;
  437. ndependson = 0;
  438. tag = -1;
  439. workerid = -1;
  440. footprint = 0;
  441. startTime = 0.0;
  442. endTime = 0.0;
  443. if (submitorder != -1)
  444. submitorder = -1;
  445. iteration = -1;
  446. nb_parameters = 0;
  447. alloc_mode = 1;
  448. }
  449. void fix_wontuse_handle(struct task * wontuseTask)
  450. {
  451. STARPU_ASSERT(wontuseTask);
  452. if (!wontuseTask->reg_signal)
  453. /* Data was already registered when we created this task, so it's already a handle */
  454. return;
  455. struct handle *handle_tmp;
  456. /* Data was not registered when we created this task, so this is the application pointer, look it up now */
  457. HASH_FIND(hh, handles_hash, &wontuseTask->task.handles[0], sizeof(wontuseTask->task.handles[0]), handle_tmp);
  458. if (handle_tmp)
  459. wontuseTask->task.handles[0] = handle_tmp->mem_ptr;
  460. else
  461. /* This data wasn't actually used, don't care about it */
  462. wontuseTask->task.handles[0] = NULL;
  463. }
  464. /* Function that submits all the tasks (used when the program reaches EOF) */
  465. int submit_tasks(void)
  466. {
  467. /* Add dependencies */
  468. const struct starpu_rbtree * tmptree = &tree;
  469. struct starpu_rbtree_node * currentNode = starpu_rbtree_first(tmptree);
  470. long last_submitorder = 0;
  471. while (currentNode != NULL)
  472. {
  473. struct task * currentTask = (struct task *) currentNode;
  474. if (currentTask->type == NormalTask)
  475. {
  476. if (currentTask->submit_order != -1)
  477. {
  478. STARPU_ASSERT(currentTask->submit_order >= last_submitorder + 1);
  479. while (currentTask->submit_order > last_submitorder + 1)
  480. {
  481. /* Oops, some tasks were not submitted by original application, fake some */
  482. struct starpu_task *task = starpu_task_create();
  483. int ret;
  484. task->cl = NULL;
  485. ret = starpu_task_submit(task);
  486. STARPU_ASSERT(ret == 0);
  487. last_submitorder++;
  488. }
  489. }
  490. if (currentTask->ndependson > 0)
  491. {
  492. struct starpu_task * taskdeps[currentTask->ndependson];
  493. unsigned i, j = 0;
  494. for (i = 0; i < currentTask->ndependson; i++)
  495. {
  496. struct task * taskdep;
  497. /* Get the ith jobid of deps_jobid */
  498. HASH_FIND(hh, tasks, &currentTask->deps[i], sizeof(jobid), taskdep);
  499. if(taskdep)
  500. {
  501. taskdeps[j] = &taskdep->task;
  502. j ++;
  503. }
  504. }
  505. starpu_task_declare_deps_array(&currentTask->task, j, taskdeps);
  506. }
  507. if (!(currentTask->iteration == -1))
  508. starpu_iteration_push(currentTask->iteration);
  509. applySchedRec(&currentTask->task, currentTask->submit_order);
  510. int ret_val = starpu_task_submit(&currentTask->task);
  511. if (!(currentTask->iteration == -1))
  512. starpu_iteration_pop();
  513. if (ret_val != 0)
  514. {
  515. fprintf(stderr, "\nWhile submitting task %ld (%s): return %d\n",
  516. currentTask->submit_order,
  517. currentTask->task.name? currentTask->task.name : "unknown",
  518. ret_val);
  519. return -1;
  520. }
  521. //fprintf(stderr, "submitting task %s (%lu, %llu)\n", currentTask->task.name?currentTask->task.name:"anonymous", currentTask->jobid, (unsigned long long) currentTask->task.tag_id);
  522. if (!(currentTask->submit_order % 1000))
  523. {
  524. fprintf(stderr, "\rSubmitted task order %ld...", currentTask->submit_order);
  525. fflush(stdout);
  526. }
  527. if (currentTask->submit_order != -1)
  528. last_submitorder++;
  529. }
  530. else
  531. {
  532. fix_wontuse_handle(currentTask); /* Add the handle in the wontuse task */
  533. if (currentTask->task.handles[0])
  534. starpu_data_wont_use(currentTask->task.handles[0]);
  535. }
  536. currentNode = starpu_rbtree_next(currentNode);
  537. }
  538. fprintf(stderr, " done.\n");
  539. return 1;
  540. }
  541. /* * * * * * * * * * * * * * * */
  542. /* * * * * * MAIN * * * * * * */
  543. /* * * * * * * * * * * * * * */
  544. static void usage(const char *program)
  545. {
  546. fprintf(stderr,"Usage: %s [--static-workerid] tasks.rec [sched.rec]\n", program);
  547. exit(EXIT_FAILURE);
  548. }
  549. int main(int argc, char **argv)
  550. {
  551. FILE *rec;
  552. char *s;
  553. const char *tasks_rec = NULL;
  554. const char *sched_rec = NULL;
  555. unsigned i;
  556. size_t s_allocated = 128;
  557. unsigned long nread_tasks = 0;
  558. /* FIXME: we do not support data with sequential consistency disabled */
  559. _STARPU_MALLOC(s, s_allocated);
  560. dependson_size = REPLAY_NMAX_DEPENDENCIES; /* Change the value of REPLAY_NMAX_DEPENCIES to modify the number of dependencies */
  561. _STARPU_MALLOC(dependson, dependson_size * sizeof (* dependson));
  562. alloc_mode = 1;
  563. for (i = 1; i < (unsigned) argc; i++)
  564. {
  565. if (!strcmp(argv[i], "--help") || !strcmp(argv[i], "-h"))
  566. {
  567. usage(argv[0]);
  568. }
  569. else if (!strcmp(argv[i], "--static-workerid"))
  570. {
  571. static_workerid = 1;
  572. }
  573. else
  574. {
  575. if (!tasks_rec)
  576. tasks_rec = argv[i];
  577. else if (!sched_rec)
  578. sched_rec = argv[i];
  579. else
  580. usage(argv[0]);
  581. }
  582. }
  583. if (!tasks_rec)
  584. usage(argv[0]);
  585. if (sched_rec)
  586. schedRecInit(sched_rec);
  587. rec = fopen(tasks_rec, "r");
  588. if (!rec)
  589. {
  590. fprintf(stderr,"unable to open file %s: %s\n", tasks_rec, strerror(errno));
  591. exit(EXIT_FAILURE);
  592. }
  593. int ret = starpu_init(NULL);
  594. if (ret == -ENODEV) goto enodev;
  595. /* Read line by line, and on empty line submit the task with the accumulated information */
  596. reset();
  597. double start = starpu_timing_now();
  598. int linenum = 0;
  599. while(1)
  600. {
  601. char *ln;
  602. if (!fgets(s, s_allocated, rec))
  603. {
  604. fprintf(stderr, " done.\n");
  605. int submitted = submit_tasks();
  606. if (submitted == -1)
  607. {
  608. goto enodev;
  609. }
  610. goto eof;
  611. }
  612. while (!(ln = strchr(s, '\n')))
  613. {
  614. /* fprintf(stderr,"buffer size %d too small, doubling it\n", s_allocated); */
  615. _STARPU_REALLOC(s, s_allocated * 2);
  616. if (!fgets(s + s_allocated-1, s_allocated+1, rec))
  617. {
  618. fprintf(stderr, "\n");
  619. int submitted = submit_tasks();
  620. if (submitted == -1)
  621. {
  622. goto enodev;
  623. }
  624. goto eof;
  625. }
  626. s_allocated *= 2;
  627. }
  628. linenum++;
  629. if (ln == s)
  630. {
  631. /* Empty line, do task */
  632. struct task * task;
  633. _STARPU_MALLOC(task, sizeof(*task));
  634. starpu_task_init(&task->task);
  635. task->deps = NULL;
  636. task->submit_order = submitorder;
  637. starpu_rbtree_node_init(&task->node);
  638. starpu_rbtree_insert(&tree, &task->node, diff);
  639. task->jobid = jobid;
  640. task->iteration = iteration;
  641. if (name != NULL)
  642. task->task.name = strdup(name);
  643. task->type = control;
  644. if (control == NormalTask)
  645. {
  646. if (workerid >= 0)
  647. {
  648. task->task.priority = priority;
  649. task->task.cl = &cl;
  650. if (static_workerid)
  651. {
  652. task->task.workerid = workerid;
  653. task->task.execute_on_a_specific_worker = 1;
  654. }
  655. if (alloc_mode)
  656. {
  657. /* Duplicating the handles stored (and registered in the current context) into the task */
  658. ARRAY_DUP(modes_ptr, task->task.modes, nb_parameters);
  659. ARRAY_DUP(modes_ptr, task->task.cl->modes, nb_parameters);
  660. variable_data_register_check(sizes_set, nb_parameters);
  661. ARRAY_DUP(handles_ptr, task->task.handles, nb_parameters);
  662. }
  663. else
  664. {
  665. task->task.dyn_modes = modes_ptr;
  666. _STARPU_MALLOC(task->task.cl->dyn_modes, (sizeof(*task->task.cl->dyn_modes) * nb_parameters));
  667. ARRAY_DUP(modes_ptr, task->task.cl->dyn_modes, nb_parameters);
  668. variable_data_register_check(sizes_set, nb_parameters);
  669. task->task.dyn_handles = handles_ptr;
  670. }
  671. task->task.nbuffers = nb_parameters;
  672. struct perfmodel * realmodel;
  673. HASH_FIND_STR(model_hash, model, realmodel);
  674. if (realmodel == NULL)
  675. {
  676. int len = strlen(model);
  677. _STARPU_CALLOC(realmodel, 1, sizeof(struct perfmodel));
  678. _STARPU_MALLOC(realmodel->model_name, sizeof(char) * (len+1));
  679. realmodel->model_name = strcpy(realmodel->model_name, model);
  680. starpu_perfmodel_init(&realmodel->perfmodel);
  681. int error = starpu_perfmodel_load_symbol(model, &realmodel->perfmodel);
  682. if (!error)
  683. {
  684. HASH_ADD_STR(model_hash, model_name, realmodel);
  685. }
  686. else
  687. {
  688. fprintf(stderr, "[starpu][Warning] Error loading perfmodel symbol %s\n", model);
  689. fprintf(stderr, "[starpu][Warning] Taking only measurements from the given execution, and forcing execution on worker %d\n", workerid);
  690. starpu_perfmodel_unload_model(&realmodel->perfmodel);
  691. free(realmodel->model_name);
  692. free(realmodel);
  693. realmodel = NULL;
  694. }
  695. }
  696. struct starpu_perfmodel_arch *arch = starpu_worker_get_perf_archtype(workerid, 0);
  697. unsigned comb = starpu_perfmodel_arch_comb_add(arch->ndevices, arch->devices);
  698. unsigned narch = starpu_perfmodel_get_narch_combs();
  699. struct task_arg *arg;
  700. _STARPU_MALLOC(arg, sizeof(struct task_arg) + sizeof(double) * narch);
  701. arg->footprint = footprint;
  702. arg->narch = narch;
  703. double * perfTime = arg->perf;
  704. if (realmodel == NULL)
  705. {
  706. /* Erf, do without perfmodel, for execution there */
  707. task->task.workerid = workerid;
  708. task->task.execute_on_a_specific_worker = 1;
  709. for (i = 0; i < narch ; i++)
  710. {
  711. if (i == comb)
  712. perfTime[i] = endTime - startTime;
  713. else
  714. perfTime[i] = NAN;
  715. }
  716. }
  717. else
  718. {
  719. int one = 0;
  720. for (i = 0; i < narch ; i++)
  721. {
  722. arch = starpu_perfmodel_arch_comb_fetch(i);
  723. perfTime[i] = starpu_perfmodel_history_based_expected_perf(&realmodel->perfmodel, arch, footprint);
  724. if (!(perfTime[i] == 0 || isnan(perfTime[i])))
  725. one = 1;
  726. }
  727. if (!one)
  728. {
  729. fprintf(stderr, "We do not have any performance measurement for symbol '%s' for footprint %x, we can not execute this", model, footprint);
  730. exit(EXIT_FAILURE);
  731. }
  732. }
  733. task->task.cl_arg = arg;
  734. task->task.flops = flops;
  735. total_flops += flops;
  736. }
  737. task->task.cl_arg_size = 0;
  738. task->task.tag_id = tag;
  739. task->task.use_tag = 1;
  740. task->ndependson = ndependson;
  741. if (ndependson > 0)
  742. {
  743. _STARPU_MALLOC(task->deps, ndependson * sizeof (* task->deps));
  744. ARRAY_DUP(dependson, task->deps, ndependson);
  745. }
  746. }
  747. else
  748. {
  749. STARPU_ASSERT(nb_parameters == 1);
  750. task->reg_signal = reg_signal[0];
  751. ARRAY_DUP(handles_ptr, task->task.handles, nb_parameters);
  752. }
  753. /* Add this task to task hash */
  754. HASH_ADD(hh, tasks, jobid, sizeof(jobid), task);
  755. nread_tasks++;
  756. if (!(nread_tasks % 1000))
  757. {
  758. fprintf(stderr, "\rRead task %lu...", nread_tasks);
  759. fflush(stdout);
  760. }
  761. reset();
  762. }
  763. /* Record various information */
  764. #define TEST(field) (!strncmp(s, field": ", strlen(field) + 2))
  765. else if(TEST("Control"))
  766. {
  767. char * c = s+9;
  768. if(!strncmp(c, "WontUse", 7))
  769. {
  770. control = WontUseTask;
  771. nb_parameters = 1;
  772. alloc_mode = set_alloc_mode(nb_parameters);
  773. arrays_managing(alloc_mode);
  774. }
  775. else
  776. control = NormalTask;
  777. }
  778. else if (TEST("Name"))
  779. {
  780. *ln = 0;
  781. name = strdup(s+6);
  782. }
  783. else if (TEST("Model"))
  784. {
  785. *ln = 0;
  786. model = strdup(s+7);
  787. }
  788. else if (TEST("JobId"))
  789. jobid = atol(s+7);
  790. else if(TEST("SubmitOrder"))
  791. submitorder = atoi(s+13);
  792. else if (TEST("DependsOn"))
  793. {
  794. char *c = s + 11;
  795. for (ndependson = 0; *c != '\n'; ndependson++)
  796. {
  797. if (ndependson >= dependson_size)
  798. {
  799. dependson_size *= 2;
  800. _STARPU_REALLOC(dependson, dependson_size * sizeof(*dependson));
  801. }
  802. dependson[ndependson] = strtol(c, &c, 10);
  803. }
  804. }
  805. else if (TEST("Tag"))
  806. {
  807. tag = strtol(s+5, NULL, 16);
  808. }
  809. else if (TEST("WorkerId"))
  810. {
  811. workerid = atoi(s+10);
  812. }
  813. else if (TEST("Footprint"))
  814. {
  815. footprint = strtoul(s+11, NULL, 16);
  816. }
  817. else if (TEST("Parameters"))
  818. {
  819. /* Nothing to do */
  820. }
  821. else if (TEST("Handles"))
  822. {
  823. *ln = 0;
  824. char *buffer = s + 9;
  825. const char *delim = " ";
  826. unsigned nb_parameters_line = count_number_tokens(buffer, delim);
  827. if(nb_parameters == 0)
  828. {
  829. nb_parameters = nb_parameters_line;
  830. arrays_managing(set_alloc_mode(nb_parameters));
  831. }
  832. else
  833. STARPU_ASSERT(nb_parameters == nb_parameters_line);
  834. char* token = strtok(buffer, delim);
  835. for (i = 0 ; i < nb_parameters ; i++)
  836. {
  837. STARPU_ASSERT(token);
  838. struct handle *handles_cell; /* A cell of the hash table for the handles */
  839. starpu_data_handle_t handle_value = (starpu_data_handle_t) strtol(token, NULL, 16); /* Get the ith handle on the line (in the file) */
  840. HASH_FIND(hh, handles_hash, &handle_value, sizeof(handle_value), handles_cell); /* Find if the handle_value was already registered as a key in the hash table */
  841. /* If it wasn't, then add it to the hash table */
  842. if (handles_cell == NULL)
  843. {
  844. /* Hide the initial handle from the file into the handles array to find it when necessary */
  845. handles_ptr[i] = handle_value;
  846. reg_signal[i] = 1;
  847. }
  848. else
  849. {
  850. handles_ptr[i] = handles_cell->mem_ptr;
  851. reg_signal[i] = 0;
  852. }
  853. token = strtok(NULL, delim);
  854. }
  855. }
  856. else if (TEST("Modes"))
  857. {
  858. *ln = 0;
  859. char * buffer = s + 7;
  860. unsigned mode_i = 0;
  861. const char * delim = " ";
  862. unsigned nb_parameters_line = count_number_tokens(buffer, delim);
  863. if(nb_parameters == 0)
  864. {
  865. nb_parameters = nb_parameters_line;
  866. arrays_managing(set_alloc_mode(nb_parameters));
  867. }
  868. else
  869. STARPU_ASSERT(nb_parameters == nb_parameters_line);
  870. char* token = strtok(buffer, delim);
  871. while (token != NULL && mode_i < nb_parameters)
  872. {
  873. /* Subject to the names of starpu modes enumerator are not modified */
  874. if (!strncmp(token, "RW", 2))
  875. {
  876. *(modes_ptr+mode_i) = STARPU_RW;
  877. mode_i++;
  878. }
  879. else if (!strncmp(token, "R", 1))
  880. {
  881. *(modes_ptr+mode_i) = STARPU_R;
  882. mode_i++;
  883. }
  884. else if (!strncmp(token, "W", 1))
  885. {
  886. *(modes_ptr+mode_i) = STARPU_W;
  887. mode_i++;
  888. }
  889. /* Other cases produce a warning*/
  890. else
  891. {
  892. fprintf(stderr, "[Warning] A mode is different from R/W (jobid task : %lu)", jobid);
  893. }
  894. token = strtok(NULL, delim);
  895. }
  896. }
  897. else if (TEST("Sizes"))
  898. {
  899. *ln = 0;
  900. char * buffer = s + 7;
  901. const char * delim = " ";
  902. unsigned nb_parameters_line = count_number_tokens(buffer, delim);
  903. unsigned k = 0;
  904. if(nb_parameters == 0)
  905. {
  906. nb_parameters = nb_parameters_line;
  907. arrays_managing(set_alloc_mode(nb_parameters));
  908. }
  909. else
  910. STARPU_ASSERT(nb_parameters == nb_parameters_line);
  911. _STARPU_MALLOC(sizes_set, nb_parameters * sizeof(size_t));
  912. char * token = strtok(buffer, delim);
  913. while (token != NULL && k < nb_parameters)
  914. {
  915. sizes_set[k] = strtol(token, NULL, 10);
  916. token = strtok(NULL, delim);
  917. k++;
  918. }
  919. }
  920. else if (TEST("StartTime"))
  921. {
  922. startTime = strtod(s+11, NULL);
  923. }
  924. else if (TEST("EndTime"))
  925. {
  926. endTime = strtod(s+9, NULL);
  927. }
  928. else if (TEST("GFlop"))
  929. {
  930. flops = 1000000000 * strtod(s+7, NULL);
  931. }
  932. else if (TEST("Iteration"))
  933. {
  934. iteration = (unsigned) strtol(s+11, NULL, 10);
  935. }
  936. else if (TEST("Priority"))
  937. {
  938. priority = strtol(s + 10, NULL, 10);
  939. }
  940. }
  941. eof:
  942. starpu_task_wait_for_all();
  943. fprintf(stderr, " done.\n");
  944. printf("%g ms", (starpu_timing_now() - start) / 1000.);
  945. if (total_flops != 0.)
  946. printf("\t%g GF/s", (total_flops / (starpu_timing_now() - start)) / 1000.);
  947. printf("\n");
  948. /* FREE allocated memory */
  949. free(dependson);
  950. free(s);
  951. /* End of FREE */
  952. struct handle *handle=NULL, *handletmp=NULL;
  953. HASH_ITER(hh, handles_hash, handle, handletmp)
  954. {
  955. starpu_data_unregister(handle->mem_ptr);
  956. HASH_DEL(handles_hash, handle);
  957. free(handle);
  958. }
  959. struct perfmodel *model_s=NULL, *modeltmp=NULL;
  960. HASH_ITER(hh, model_hash, model_s, modeltmp)
  961. {
  962. starpu_perfmodel_unload_model(&model_s->perfmodel);
  963. HASH_DEL(model_hash, model_s);
  964. free(model_s->model_name);
  965. free(model_s);
  966. }
  967. struct task *task=NULL, *tasktmp=NULL;
  968. HASH_ITER(hh, tasks, task, tasktmp)
  969. {
  970. free(task->task.cl_arg);
  971. free((char*)task->task.name);
  972. if (task->task.dyn_handles != NULL)
  973. {
  974. free(task->task.dyn_handles);
  975. free(task->task.dyn_modes);
  976. }
  977. HASH_DEL(tasks, task);
  978. starpu_task_clean(&task->task);
  979. free(task->deps);
  980. starpu_rbtree_remove(&tree, &task->node);
  981. free(task);
  982. }
  983. starpu_shutdown();
  984. return 0;
  985. enodev:
  986. starpu_shutdown();
  987. return 77;
  988. }