starpu_replay.c 30 KB

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