simgrid.c 41 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2012-2021 Université de Bordeaux, CNRS (LaBRI UMR 5800), Inria
  4. * Copyright (C) 2013 Thibaut Lambert
  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 <datawizard/memory_nodes.h>
  19. #include <common/config.h>
  20. #ifdef HAVE_UNISTD_H
  21. #include <unistd.h>
  22. #endif
  23. #include <core/perfmodel/perfmodel.h>
  24. #include <core/workers.h>
  25. #include <core/simgrid.h>
  26. #if defined(HAVE_SIMGRID_SIMDAG_H) && (SIMGRID_VERSION >= 31300)
  27. #include <simgrid/simdag.h>
  28. #endif
  29. #ifdef STARPU_SIMGRID
  30. #ifdef HAVE_GETRLIMIT
  31. #include <sys/resource.h>
  32. #endif
  33. #include <simgrid/simix.h>
  34. #ifdef STARPU_HAVE_SIMGRID_HOST_H
  35. #include <simgrid/host.h>
  36. #endif
  37. #ifdef STARPU_HAVE_SIMGRID_ENGINE_H
  38. #include <simgrid/engine.h>
  39. #endif
  40. #ifdef STARPU_HAVE_XBT_CONFIG_H
  41. #include <xbt/config.h>
  42. #endif
  43. #include <smpi/smpi.h>
  44. #pragma weak starpu_main
  45. extern int starpu_main(int argc, char *argv[]);
  46. #if SIMGRID_VERSION < 31600
  47. #pragma weak smpi_main
  48. extern int smpi_main(int (*realmain) (int argc, char *argv[]), int argc, char *argv[]);
  49. #endif
  50. #pragma weak _starpu_mpi_simgrid_init
  51. extern int _starpu_mpi_simgrid_init(int argc, char *argv[]);
  52. #pragma weak smpi_process_set_user_data
  53. #if !HAVE_DECL_SMPI_PROCESS_SET_USER_DATA && !defined(smpi_process_set_user_data)
  54. extern void smpi_process_set_user_data(void *);
  55. #endif
  56. static double _starpu_simgrid_dynamic_energy = 0.0;
  57. /* 1 when MSG_init was done, 2 when initialized through redirected main, 3 when
  58. * initialized through MSG_process_attach */
  59. static int simgrid_started;
  60. static int simgrid_transfer_cost = 1;
  61. static int runners_running;
  62. starpu_pthread_queue_t _starpu_simgrid_transfer_queue[STARPU_MAXNODES];
  63. static struct transfer_runner
  64. {
  65. struct transfer *first_transfer, *last_transfer;
  66. starpu_sem_t sem;
  67. starpu_pthread_t runner;
  68. } transfer_runner[STARPU_MAXNODES][STARPU_MAXNODES];
  69. static void *transfer_execute(void *arg);
  70. starpu_pthread_queue_t _starpu_simgrid_task_queue[STARPU_NMAXWORKERS];
  71. static struct worker_runner
  72. {
  73. struct task *first_task, *last_task;
  74. starpu_sem_t sem;
  75. starpu_pthread_t runner;
  76. } worker_runner[STARPU_NMAXWORKERS];
  77. static void *task_execute(void *arg);
  78. size_t _starpu_default_stack_size = 8192;
  79. void _starpu_simgrid_set_stack_size(size_t stack_size)
  80. {
  81. #ifdef HAVE_SG_CFG_SET_INT
  82. sg_cfg_set_int("contexts/stack-size", stack_size);
  83. #elif SIMGRID_VERSION >= 31300
  84. xbt_cfg_set_int("contexts/stack-size", stack_size);
  85. #else
  86. extern xbt_cfg_t _sg_cfg_set;
  87. xbt_cfg_set_int(_sg_cfg_set, "contexts/stack_size", stack_size);
  88. #endif
  89. }
  90. #ifdef HAVE_SG_ACTOR_ON_EXIT
  91. static void on_exit_backtrace(int failed, void *data STARPU_ATTRIBUTE_UNUSED)
  92. {
  93. if (failed)
  94. xbt_backtrace_display_current();
  95. }
  96. #endif
  97. void _starpu_simgrid_actor_setup(void)
  98. {
  99. #ifdef HAVE_SG_ACTOR_ON_EXIT
  100. sg_actor_on_exit(on_exit_backtrace, NULL);
  101. #endif
  102. }
  103. #if defined(HAVE_SG_ZONE_GET_BY_NAME) || defined(sg_zone_get_by_name)
  104. #define HAVE_STARPU_SIMGRID_GET_AS_BY_NAME
  105. sg_netzone_t _starpu_simgrid_get_as_by_name(const char *name)
  106. {
  107. return sg_zone_get_by_name(name);
  108. }
  109. #elif defined(HAVE_MSG_ZONE_GET_BY_NAME) || defined(MSG_zone_get_by_name)
  110. #define HAVE_STARPU_SIMGRID_GET_AS_BY_NAME
  111. msg_as_t _starpu_simgrid_get_as_by_name(const char *name)
  112. {
  113. return MSG_zone_get_by_name(name);
  114. }
  115. #elif defined(HAVE_MSG_GET_AS_BY_NAME) || defined(MSG_get_as_by_name)
  116. #define HAVE_STARPU_SIMGRID_GET_AS_BY_NAME
  117. msg_as_t _starpu_simgrid_get_as_by_name(const char *name)
  118. {
  119. return MSG_get_as_by_name(name);
  120. }
  121. #elif defined(HAVE_MSG_ENVIRONMENT_GET_ROUTING_ROOT) || defined(MSG_environment_as_get_routing_sons)
  122. #define HAVE_STARPU_SIMGRID_GET_AS_BY_NAME
  123. static msg_as_t __starpu_simgrid_get_as_by_name(msg_as_t root, const char *name)
  124. {
  125. xbt_dict_t dict;
  126. xbt_dict_cursor_t cursor;
  127. const char *key;
  128. msg_as_t as, ret;
  129. dict = MSG_environment_as_get_routing_sons(root);
  130. xbt_dict_foreach(dict, cursor, key, as)
  131. {
  132. if (!strcmp(MSG_environment_as_get_name(as), name))
  133. return as;
  134. ret = __starpu_simgrid_get_as_by_name(as, name);
  135. if (ret)
  136. return ret;
  137. }
  138. return NULL;
  139. }
  140. msg_as_t _starpu_simgrid_get_as_by_name(const char *name)
  141. {
  142. return __starpu_simgrid_get_as_by_name(MSG_environment_get_routing_root(), name);
  143. }
  144. #endif /* HAVE_MSG_ENVIRONMENT_GET_ROUTING_ROOT */
  145. int _starpu_simgrid_get_nbhosts(const char *prefix)
  146. {
  147. int ret;
  148. #ifdef HAVE_SG_HOST_LIST
  149. sg_host_t *hosts_list = NULL;
  150. #endif
  151. xbt_dynar_t hosts = NULL;
  152. unsigned i, nb = 0;
  153. unsigned len = strlen(prefix);
  154. if (_starpu_simgrid_running_smpi())
  155. {
  156. #ifdef HAVE_STARPU_SIMGRID_GET_AS_BY_NAME
  157. char new_prefix[32];
  158. char name[32];
  159. STARPU_ASSERT(starpu_mpi_world_rank);
  160. snprintf(name, sizeof(name), STARPU_MPI_AS_PREFIX"%d", starpu_mpi_world_rank());
  161. #if defined(HAVE_MSG_ZONE_GET_HOSTS) || defined(HAVE_SG_ZONE_GET_HOSTS) || defined(MSG_zone_get_hosts) || defined(sg_zone_get_hosts)
  162. hosts = xbt_dynar_new(sizeof(sg_host_t), NULL);
  163. # if defined(HAVE_SG_ZONE_GET_HOSTS) || defined(sg_zone_get_hosts)
  164. sg_zone_get_hosts(_starpu_simgrid_get_as_by_name(name), hosts);
  165. # else
  166. MSG_zone_get_hosts(_starpu_simgrid_get_as_by_name(name), hosts);
  167. # endif
  168. #else
  169. hosts = MSG_environment_as_get_hosts(_starpu_simgrid_get_as_by_name(name));
  170. #endif
  171. snprintf(new_prefix, sizeof(new_prefix), "%s-%s", name, prefix);
  172. prefix = new_prefix;
  173. len = strlen(prefix);
  174. #else
  175. STARPU_ABORT_MSG("can not continue without an implementation for _starpu_simgrid_get_as_by_name");
  176. #endif /* HAVE_STARPU_SIMGRID_GET_AS_BY_NAME */
  177. }
  178. else
  179. {
  180. #ifdef HAVE_SG_HOST_LIST
  181. hosts_list = sg_host_list();
  182. nb = sg_host_count();
  183. #elif defined(STARPU_HAVE_SIMGRID_HOST_H)
  184. hosts = sg_hosts_as_dynar();
  185. #else
  186. hosts = MSG_hosts_as_dynar();
  187. #endif
  188. }
  189. if (hosts)
  190. nb = xbt_dynar_length(hosts);
  191. ret = 0;
  192. for (i = 0; i < nb; i++)
  193. {
  194. const char *name;
  195. #ifdef HAVE_SG_HOST_LIST
  196. if (hosts_list)
  197. name = sg_host_get_name(hosts_list[i]);
  198. else
  199. #endif
  200. #if defined(STARPU_HAVE_SIMGRID_HOST_H)
  201. name = sg_host_get_name(xbt_dynar_get_as(hosts, i, sg_host_t));
  202. #else
  203. name = MSG_host_get_name(xbt_dynar_get_as(hosts, i, msg_host_t));
  204. #endif
  205. if (!strncmp(name, prefix, len))
  206. ret++;
  207. }
  208. if (hosts)
  209. xbt_dynar_free(&hosts);
  210. return ret;
  211. }
  212. unsigned long long _starpu_simgrid_get_memsize(const char *prefix, unsigned devid)
  213. {
  214. char name[32];
  215. starpu_sg_host_t host;
  216. const char *memsize;
  217. snprintf(name, sizeof(name), "%s%u", prefix, devid);
  218. host = _starpu_simgrid_get_host_by_name(name);
  219. if (!host)
  220. return 0;
  221. #ifdef HAVE_SG_HOST_GET_PROPERTIES
  222. if (!sg_host_get_properties(host))
  223. #else
  224. if (!MSG_host_get_properties(host))
  225. #endif
  226. return 0;
  227. #ifdef HAVE_SG_HOST_GET_PROPERTIES
  228. memsize = sg_host_get_property_value(host, "memsize");
  229. #else
  230. memsize = MSG_host_get_property_value(host, "memsize");
  231. #endif
  232. if (!memsize)
  233. return 0;
  234. return atoll(memsize);
  235. }
  236. starpu_sg_host_t _starpu_simgrid_get_host_by_name(const char *name)
  237. {
  238. if (_starpu_simgrid_running_smpi())
  239. {
  240. char mpiname[32];
  241. STARPU_ASSERT(starpu_mpi_world_rank);
  242. snprintf(mpiname, sizeof(mpiname), STARPU_MPI_AS_PREFIX"%d-%s", starpu_mpi_world_rank(), name);
  243. #ifdef STARPU_HAVE_SIMGRID_HOST_H
  244. return sg_host_by_name(mpiname);
  245. #else
  246. return MSG_get_host_by_name(mpiname);
  247. #endif
  248. }
  249. else
  250. #ifdef STARPU_HAVE_SIMGRID_HOST_H
  251. return sg_host_by_name(name);
  252. #else
  253. return MSG_get_host_by_name(name);
  254. #endif
  255. }
  256. starpu_sg_host_t _starpu_simgrid_get_host_by_worker(struct _starpu_worker *worker)
  257. {
  258. char *prefix;
  259. char name[16];
  260. starpu_sg_host_t host;
  261. switch (worker->arch)
  262. {
  263. case STARPU_CPU_WORKER:
  264. prefix = "CPU";
  265. break;
  266. case STARPU_CUDA_WORKER:
  267. prefix = "CUDA";
  268. break;
  269. case STARPU_OPENCL_WORKER:
  270. prefix = "OpenCL";
  271. break;
  272. default:
  273. STARPU_ASSERT(0);
  274. }
  275. snprintf(name, sizeof(name), "%s%u", prefix, worker->devid);
  276. host = _starpu_simgrid_get_host_by_name(name);
  277. STARPU_ASSERT_MSG(host, "Could not find host %s!", name);
  278. return host;
  279. }
  280. /* Simgrid up to 3.15 would rename main into smpi_simulated_main_, and call that
  281. * from SMPI initialization
  282. * In case the MPI application didn't use smpicc to build the file containing
  283. * main(), but included our #define main starpu_main, try to cope by calling
  284. * starpu_main */
  285. int _starpu_smpi_simulated_main_(int argc, char *argv[])
  286. {
  287. if (!starpu_main)
  288. {
  289. _STARPU_ERROR("In simgrid mode, the file containing the main() function of this application needs to be compiled with starpu.h or starpu_simgrid_wrap.h included, to properly rename it into starpu_main\n");
  290. }
  291. return starpu_main(argc, argv);
  292. }
  293. int smpi_simulated_main_(int argc, char *argv[]) __attribute__((weak, alias("_starpu_smpi_simulated_main_")));
  294. /* This is used to start a non-MPI simgrid environment */
  295. void _starpu_start_simgrid(int *argc, char **argv)
  296. {
  297. char path[256];
  298. if (simgrid_started)
  299. return;
  300. simgrid_started = 1;
  301. #if defined(STARPU_SIMGRID_HAVE_SIMGRID_INIT) && defined(HAVE_SG_ACTOR_INIT)
  302. simgrid_init(argc, argv);
  303. #else
  304. MSG_init(argc, argv);
  305. #endif
  306. /* Simgrid uses tiny stacks by default. This comes unexpected to our users. */
  307. #ifdef HAVE_GETRLIMIT
  308. struct rlimit rlim;
  309. if (getrlimit(RLIMIT_STACK, &rlim) == 0 && rlim.rlim_cur != 0 && rlim.rlim_cur != RLIM_INFINITY)
  310. _starpu_default_stack_size = rlim.rlim_cur / 1024;
  311. #endif
  312. _starpu_simgrid_set_stack_size(_starpu_default_stack_size);
  313. /* Load XML platform */
  314. #if SIMGRID_VERSION < 31300
  315. _starpu_simgrid_get_platform_path(3, path, sizeof(path));
  316. #else
  317. _starpu_simgrid_get_platform_path(4, path, sizeof(path));
  318. #endif
  319. #if defined(STARPU_SIMGRID_HAVE_SIMGRID_INIT) && defined(HAVE_SG_ACTOR_INIT)
  320. simgrid_load_platform(path);
  321. #else
  322. MSG_create_environment(path);
  323. #endif
  324. int limit_bandwidth = starpu_get_env_number("STARPU_LIMIT_BANDWIDTH");
  325. if (limit_bandwidth >= 0)
  326. {
  327. #if defined(HAVE_SG_LINK_BANDWIDTH_SET) || defined(HAVE_SG_LINK_SET_BANDWIDTH)
  328. sg_link_t *links = sg_link_list();
  329. int count = sg_link_count(), i;
  330. for (i = 0; i < count; i++)
  331. {
  332. #ifdef HAVE_SG_LINK_SET_BANDWIDTH
  333. sg_link_set_bandwidth(links[i], limit_bandwidth * 1000000.);
  334. #else
  335. sg_link_bandwidth_set(links[i], limit_bandwidth * 1000000.);
  336. #endif
  337. }
  338. #else
  339. _STARPU_DISP("Warning: STARPU_LIMIT_BANDWIDTH set to %d but this requires simgrid 3.26, thus ignored\n", limit_bandwidth);
  340. #endif
  341. }
  342. simgrid_transfer_cost = starpu_get_env_number_default("STARPU_SIMGRID_TRANSFER_COST", 1);
  343. }
  344. static int main_ret;
  345. int do_starpu_main(int argc, char *argv[])
  346. {
  347. /* FIXME: Ugly work-around for bug in simgrid: the MPI context is not properly set at MSG process startup */
  348. starpu_sleep(0.000001);
  349. _starpu_simgrid_actor_setup();
  350. if (!starpu_main)
  351. {
  352. _STARPU_ERROR("In simgrid mode, the file containing the main() function of this application needs to be compiled with starpu.h or starpu_simgrid_wrap.h included, to properly rename it into starpu_main\n");
  353. }
  354. main_ret = starpu_main(argc, argv);
  355. return main_ret;
  356. }
  357. /* We need it only when using smpi */
  358. #pragma weak smpi_process_get_user_data
  359. extern void *smpi_process_get_user_data();
  360. /* This is hopefully called before the application and simgrid */
  361. #undef main
  362. #pragma weak main
  363. int main(int argc, char **argv)
  364. {
  365. #ifdef HAVE_SG_CONFIG_CONTINUE_AFTER_HELP
  366. sg_config_continue_after_help();
  367. #endif
  368. if (_starpu_simgrid_running_smpi())
  369. {
  370. if (!smpi_process_get_user_data)
  371. {
  372. _STARPU_ERROR("Your version of simgrid does not provide smpi_process_get_user_data, we can not continue without it\n");
  373. }
  374. #if SIMGRID_VERSION >= 31600
  375. /* Recent versions of simgrid dlopen() us, so we don't need to
  376. * do circumvolutions, just init MPI early and run the application's main */
  377. return _starpu_mpi_simgrid_init(argc, argv);
  378. #else
  379. /* Oops, we are running old SMPI, let it start Simgrid, and we'll
  380. * take back hand in _starpu_simgrid_init from starpu_init() */
  381. return smpi_main(_starpu_mpi_simgrid_init, argc, argv);
  382. #endif
  383. }
  384. /* Already initialized? It probably has been done through a
  385. * constructor and MSG_process_attach, directly jump to real main */
  386. if (simgrid_started == 3)
  387. {
  388. return do_starpu_main(argc, argv);
  389. }
  390. /* Managed to catch application's main, initialize simgrid first */
  391. _starpu_start_simgrid(&argc, argv);
  392. simgrid_started = 2;
  393. /* Create a simgrid process for main */
  394. char **argv_cpy;
  395. _STARPU_MALLOC(argv_cpy, argc * sizeof(char*));
  396. int i;
  397. for (i = 0; i < argc; i++)
  398. argv_cpy[i] = strdup(argv[i]);
  399. /* Run the application in a separate thread */
  400. _starpu_simgrid_actor_create("main", &do_starpu_main, _starpu_simgrid_get_host_by_name("MAIN"), argc, argv_cpy);
  401. /* And run maestro in the main thread */
  402. #if defined(STARPU_SIMGRID_HAVE_SIMGRID_INIT) && defined(HAVE_SG_ACTOR_INIT)
  403. simgrid_run();
  404. #else
  405. MSG_main();
  406. #endif
  407. return main_ret;
  408. }
  409. #if defined(HAVE_MSG_PROCESS_ATTACH) || defined(MSG_process_attach) || defined(HAVE_SG_ACTOR_ATTACH)
  410. static void maestro(void *data STARPU_ATTRIBUTE_UNUSED)
  411. {
  412. #if defined(STARPU_SIMGRID_HAVE_SIMGRID_INIT) && defined(HAVE_SG_ACTOR_INIT)
  413. simgrid_run();
  414. #else
  415. MSG_main();
  416. #endif
  417. }
  418. #endif
  419. /* This is called early from starpu_init, so thread functions etc. can work */
  420. void _starpu_simgrid_init_early(int *argc STARPU_ATTRIBUTE_UNUSED, char ***argv STARPU_ATTRIBUTE_UNUSED)
  421. {
  422. #ifdef HAVE_SG_CONFIG_CONTINUE_AFTER_HELP
  423. sg_config_continue_after_help();
  424. #endif
  425. #if defined(HAVE_MSG_PROCESS_ATTACH) || defined(MSG_process_attach) || defined(HAVE_SG_ACTOR_ATTACH)
  426. if (simgrid_started < 2 && !_starpu_simgrid_running_smpi())
  427. {
  428. /* "Cannot create_maestro with this ContextFactory.
  429. * Try using --cfg=contexts/factory:thread instead."
  430. * See https://github.com/simgrid/simgrid/issues/141 */
  431. _STARPU_DISP("Warning: In simgrid mode, the file containing the main() function of this application should to be compiled with starpu.h or starpu_simgrid_wrap.h included, to properly rename it into starpu_main to avoid having to use --cfg=contexts/factory:thread which reduces performance\n");
  432. #if SIMGRID_VERSION >= 31400 /* Only recent versions of simgrid support setting sg_cfg_set_string before starting simgrid */
  433. # ifdef HAVE_SG_CFG_SET_INT
  434. sg_cfg_set_string("contexts/factory", "thread");
  435. # else
  436. xbt_cfg_set_string("contexts/factory", "thread");
  437. # endif
  438. #endif
  439. /* We didn't catch application's main. */
  440. /* Start maestro as a separate thread */
  441. SIMIX_set_maestro(maestro, NULL);
  442. /* Initialize simgrid */
  443. _starpu_start_simgrid(argc, *argv);
  444. /* And attach the main thread to the main simgrid process */
  445. void **tsd;
  446. _STARPU_CALLOC(tsd, MAX_TSD+1, sizeof(void*));
  447. #if defined(HAVE_SG_ACTOR_ATTACH) && defined (HAVE_SG_ACTOR_DATA)
  448. sg_actor_t actor = sg_actor_attach("main", NULL, _starpu_simgrid_get_host_by_name("MAIN"), NULL);
  449. #ifdef HAVE_SG_ACTOR_SET_DATA
  450. sg_actor_set_data(actor, tsd);
  451. #else
  452. sg_actor_data_set(actor, tsd);
  453. #endif
  454. #else
  455. MSG_process_attach("main", tsd, _starpu_simgrid_get_host_by_name("MAIN"), NULL);
  456. #endif
  457. /* We initialized through MSG_process_attach */
  458. simgrid_started = 3;
  459. }
  460. #endif
  461. if (!simgrid_started && !starpu_main && !_starpu_simgrid_running_smpi())
  462. {
  463. /* Oops, we don't have MSG_process_attach and didn't catch the
  464. * 'main' symbol, there is no way for us */
  465. _STARPU_ERROR("In simgrid mode, the file containing the main() function of this application needs to be compiled with starpu.h or starpu_simgrid_wrap.h included, to properly rename it into starpu_main\n");
  466. }
  467. if (_starpu_simgrid_running_smpi())
  468. {
  469. #ifndef STARPU_STATIC_ONLY
  470. _STARPU_ERROR("Simgrid currently does not support privatization for dynamically-linked libraries in SMPI. Please reconfigure and build StarPU with --disable-shared");
  471. #endif
  472. #if defined(HAVE_MSG_PROCESS_USERDATA_INIT) && !defined(HAVE_SG_ACTOR_DATA)
  473. MSG_process_userdata_init();
  474. #endif
  475. void **tsd;
  476. _STARPU_CALLOC(tsd, MAX_TSD+1, sizeof(void*));
  477. #ifdef HAVE_SG_ACTOR_DATA
  478. #ifdef HAVE_SG_ACTOR_SET_DATA
  479. sg_actor_set_data(sg_actor_self(), tsd);
  480. #else
  481. sg_actor_data_set(sg_actor_self(), tsd);
  482. #endif
  483. #else
  484. smpi_process_set_user_data(tsd);
  485. #endif
  486. }
  487. unsigned i;
  488. for (i = 0; i < STARPU_MAXNODES; i++)
  489. starpu_pthread_queue_init(&_starpu_simgrid_transfer_queue[i]);
  490. for (i = 0; i < STARPU_NMAXWORKERS; i++)
  491. starpu_pthread_queue_init(&_starpu_simgrid_task_queue[i]);
  492. }
  493. /* This is called late from starpu_init, to start task executors */
  494. void _starpu_simgrid_init(void)
  495. {
  496. unsigned i;
  497. runners_running = 1;
  498. for (i = 0; i < starpu_worker_get_count(); i++)
  499. {
  500. char s[32];
  501. snprintf(s, sizeof(s), "worker %u runner", i);
  502. starpu_sem_init(&worker_runner[i].sem, 0, 0);
  503. starpu_pthread_create_on(s, &worker_runner[i].runner, NULL, task_execute, (void*)(uintptr_t) i, _starpu_simgrid_get_host_by_worker(_starpu_get_worker_struct(i)));
  504. }
  505. }
  506. void _starpu_simgrid_deinit_late(void)
  507. {
  508. #if defined(HAVE_MSG_PROCESS_ATTACH) || defined(MSG_process_attach) || defined(HAVE_SG_ACTOR_ATTACH)
  509. if (simgrid_started == 3)
  510. {
  511. /* Started with MSG_process_attach, now detach */
  512. #ifdef HAVE_SG_ACTOR_ATTACH
  513. sg_actor_detach();
  514. #else
  515. MSG_process_detach();
  516. #endif
  517. simgrid_started = 0;
  518. }
  519. #endif
  520. }
  521. void _starpu_simgrid_deinit(void)
  522. {
  523. unsigned i, j;
  524. runners_running = 0;
  525. for (i = 0; i < STARPU_MAXNODES; i++)
  526. {
  527. for (j = 0; j < STARPU_MAXNODES; j++)
  528. {
  529. struct transfer_runner *t = &transfer_runner[i][j];
  530. if (t->runner)
  531. {
  532. starpu_sem_post(&t->sem);
  533. #ifdef STARPU_HAVE_SIMGRID_ACTOR_H
  534. sg_actor_join(t->runner, 1000000);
  535. #elif SIMGRID_VERSION >= 31400
  536. MSG_process_join(t->runner, 1000000);
  537. #else
  538. starpu_sleep(1);
  539. #endif
  540. STARPU_ASSERT(t->first_transfer == NULL);
  541. STARPU_ASSERT(t->last_transfer == NULL);
  542. starpu_sem_destroy(&t->sem);
  543. }
  544. }
  545. /* FIXME: queue not empty at this point, needs proper unregistration */
  546. /* starpu_pthread_queue_destroy(&_starpu_simgrid_transfer_queue[i]); */
  547. }
  548. for (i = 0; i < starpu_worker_get_count(); i++)
  549. {
  550. struct worker_runner *w = &worker_runner[i];
  551. starpu_sem_post(&w->sem);
  552. #ifdef STARPU_HAVE_SIMGRID_ACTOR_H
  553. sg_actor_join(w->runner, 1000000);
  554. #elif SIMGRID_VERSION >= 31400
  555. MSG_process_join(w->runner, 1000000);
  556. #else
  557. starpu_sleep(1);
  558. #endif
  559. STARPU_ASSERT(w->first_task == NULL);
  560. STARPU_ASSERT(w->last_task == NULL);
  561. starpu_sem_destroy(&w->sem);
  562. starpu_pthread_queue_destroy(&_starpu_simgrid_task_queue[i]);
  563. }
  564. #if SIMGRID_VERSION >= 31300
  565. /* clean-atexit introduced in simgrid 3.13 */
  566. # ifdef HAVE_SG_CFG_SET_INT
  567. if ( sg_cfg_get_boolean("debug/clean-atexit"))
  568. # elif SIMGRID_VERSION >= 32300
  569. if ( xbt_cfg_get_boolean("debug/clean-atexit"))
  570. # else
  571. if ( xbt_cfg_get_boolean("clean-atexit"))
  572. # endif
  573. {
  574. _starpu_simgrid_deinit_late();
  575. }
  576. #endif
  577. }
  578. /*
  579. * Tasks
  580. */
  581. struct task
  582. {
  583. #ifdef HAVE_SG_ACTOR_SELF_EXECUTE
  584. double flops;
  585. #else
  586. msg_task_t task;
  587. #endif
  588. double energy;
  589. /* communication termination signalization */
  590. unsigned *finished;
  591. /* Next task on this worker */
  592. struct task *next;
  593. };
  594. /* Actually execute the task. */
  595. static void *task_execute(void *arg)
  596. {
  597. unsigned workerid = (uintptr_t) arg;
  598. struct worker_runner *w = &worker_runner[workerid];
  599. _STARPU_DEBUG("worker runner %u started\n", workerid);
  600. while (1)
  601. {
  602. struct task *task;
  603. starpu_sem_wait(&w->sem);
  604. if (!runners_running)
  605. break;
  606. task = w->first_task;
  607. w->first_task = task->next;
  608. if (w->last_task == task)
  609. w->last_task = NULL;
  610. _STARPU_DEBUG("task %p started\n", task);
  611. #ifdef HAVE_SG_ACTOR_EXECUTE
  612. sg_actor_execute(task->flops);
  613. #elif defined(HAVE_SG_ACTOR_SELF_EXECUTE)
  614. sg_actor_self_execute(task->flops);
  615. #else
  616. MSG_task_execute(task->task);
  617. MSG_task_destroy(task->task);
  618. #endif
  619. starpu_energy_use(task->energy);
  620. _STARPU_DEBUG("task %p finished\n", task);
  621. *task->finished = 1;
  622. /* The worker which started this task may be sleeping out of tasks, wake it */
  623. _starpu_wake_worker_relax(workerid);
  624. free(task);
  625. }
  626. _STARPU_DEBUG("worker %u stopped\n", workerid);
  627. return 0;
  628. }
  629. /* Wait for completion of all asynchronous tasks for this worker */
  630. void _starpu_simgrid_wait_tasks(int workerid)
  631. {
  632. struct task *task = worker_runner[workerid].last_task;
  633. if (!task)
  634. return;
  635. unsigned *finished = task->finished;
  636. starpu_pthread_wait_t wait;
  637. starpu_pthread_wait_init(&wait);
  638. starpu_pthread_queue_register(&wait, &_starpu_simgrid_task_queue[workerid]);
  639. while(1)
  640. {
  641. starpu_pthread_wait_reset(&wait);
  642. if (*finished)
  643. break;
  644. starpu_pthread_wait_wait(&wait);
  645. }
  646. starpu_pthread_queue_unregister(&wait, &_starpu_simgrid_task_queue[workerid]);
  647. starpu_pthread_wait_destroy(&wait);
  648. }
  649. /* Task execution submitted by StarPU */
  650. void _starpu_simgrid_submit_job(int workerid, int sched_ctx_id, struct _starpu_job *j, struct starpu_perfmodel_arch* perf_arch, double length, double energy, unsigned *finished)
  651. {
  652. struct starpu_task *starpu_task = j->task;
  653. double flops;
  654. #ifndef HAVE_SG_ACTOR_SELF_EXECUTE
  655. msg_task_t simgrid_task;
  656. #endif
  657. if (j->internal)
  658. /* This is not useful to include in simulation (and probably
  659. * doesn't have a perfmodel anyway) */
  660. return;
  661. if (isnan(length))
  662. {
  663. length = starpu_task_worker_expected_length(starpu_task, workerid, sched_ctx_id, j->nimpl);
  664. STARPU_ASSERT_MSG(!_STARPU_IS_ZERO(length) && !isnan(length),
  665. "Codelet %s does not have a perfmodel (in directory %s), or is not calibrated enough, please re-run in non-simgrid mode until it is calibrated, or fix the STARPU_HOSTNAME and STARPU_PERF_MODEL_DIR environment variables",
  666. _starpu_job_get_model_name(j), _starpu_get_perf_model_dir_codelet());
  667. /* TODO: option to add variance according to performance model,
  668. * to be able to easily check scheduling robustness */
  669. }
  670. if (isnan(energy))
  671. {
  672. energy = starpu_task_worker_expected_energy(starpu_task, workerid, sched_ctx_id, j->nimpl);
  673. /* TODO: option to add variance according to performance model,
  674. * to be able to easily check scheduling robustness */
  675. }
  676. #ifdef HAVE_SG_HOST_GET_SPEED
  677. flops = length/1000000.0*sg_host_get_speed(sg_host_self());
  678. #else
  679. #if defined(HAVE_SG_HOST_SPEED) || defined(sg_host_speed)
  680. # if defined(HAVE_SG_HOST_SELF) || defined(sg_host_self)
  681. flops = length/1000000.0*sg_host_speed(sg_host_self());
  682. # else
  683. flops = length/1000000.0*sg_host_speed(MSG_host_self());
  684. # endif
  685. #elif defined HAVE_MSG_HOST_GET_SPEED || defined(MSG_host_get_speed)
  686. flops = length/1000000.0*MSG_host_get_speed(MSG_host_self());
  687. #else
  688. flops = length/1000000.0*MSG_get_host_speed(MSG_host_self());
  689. #endif
  690. #endif
  691. #ifndef HAVE_SG_ACTOR_SELF_EXECUTE
  692. simgrid_task = MSG_task_create(_starpu_job_get_task_name(j), flops, 0, NULL);
  693. #endif
  694. if (finished == NULL)
  695. {
  696. /* Synchronous execution */
  697. /* First wait for previous tasks */
  698. _starpu_simgrid_wait_tasks(workerid);
  699. #ifdef HAVE_SG_ACTOR_EXECUTE
  700. sg_actor_execute(flops);
  701. #elif defined(HAVE_SG_ACTOR_SELF_EXECUTE)
  702. sg_actor_self_execute(flops);
  703. #else
  704. MSG_task_execute(simgrid_task);
  705. MSG_task_destroy(simgrid_task);
  706. #endif
  707. starpu_energy_use(energy);
  708. }
  709. else
  710. {
  711. /* Asynchronous execution */
  712. struct task *task;
  713. struct worker_runner *w = &worker_runner[workerid];
  714. _STARPU_MALLOC(task, sizeof(*task));
  715. #ifdef HAVE_SG_ACTOR_SELF_EXECUTE
  716. task->flops = flops;
  717. #else
  718. task->task = simgrid_task;
  719. #endif
  720. task->energy = energy;
  721. task->finished = finished;
  722. *finished = 0;
  723. task->next = NULL;
  724. /* Sleep 10µs for the GPU task queueing */
  725. if (_starpu_simgrid_queue_malloc_cost())
  726. starpu_sleep(0.000010);
  727. if (w->last_task)
  728. {
  729. /* Already running a task, queue */
  730. w->last_task->next = task;
  731. w->last_task = task;
  732. }
  733. else
  734. {
  735. STARPU_ASSERT(!w->first_task);
  736. w->first_task = task;
  737. w->last_task = task;
  738. }
  739. starpu_sem_post(&w->sem);
  740. }
  741. }
  742. /*
  743. * Transfers
  744. */
  745. /* Note: simgrid is not parallel, so there is no need to hold locks for management of transfers. */
  746. LIST_TYPE(transfer,
  747. #if defined(HAVE_SG_HOST_SEND_TO) || defined(HAVE_SG_HOST_SENDTO)
  748. size_t size;
  749. #else
  750. msg_task_t task;
  751. #endif
  752. int src_node;
  753. int dst_node;
  754. int run_node;
  755. /* communication termination signalization */
  756. unsigned *finished;
  757. /* transfers which wait for this transfer */
  758. struct transfer **wake;
  759. unsigned nwake;
  760. /* Number of transfers that this transfer waits for */
  761. unsigned nwait;
  762. /* Next transfer on this stream */
  763. struct transfer *next;
  764. )
  765. struct transfer_list pending;
  766. /* Tell for two transfers whether they should be handled in sequence */
  767. static int transfers_are_sequential(struct transfer *new_transfer, struct transfer *old_transfer)
  768. {
  769. int new_is_cuda STARPU_ATTRIBUTE_UNUSED, old_is_cuda STARPU_ATTRIBUTE_UNUSED;
  770. int new_is_opencl STARPU_ATTRIBUTE_UNUSED, old_is_opencl STARPU_ATTRIBUTE_UNUSED;
  771. int new_is_gpu_gpu, old_is_gpu_gpu;
  772. new_is_cuda = starpu_node_get_kind(new_transfer->src_node) == STARPU_CUDA_RAM;
  773. new_is_cuda |= starpu_node_get_kind(new_transfer->dst_node) == STARPU_CUDA_RAM;
  774. old_is_cuda = starpu_node_get_kind(old_transfer->src_node) == STARPU_CUDA_RAM;
  775. old_is_cuda |= starpu_node_get_kind(old_transfer->dst_node) == STARPU_CUDA_RAM;
  776. new_is_opencl = starpu_node_get_kind(new_transfer->src_node) == STARPU_OPENCL_RAM;
  777. new_is_opencl |= starpu_node_get_kind(new_transfer->dst_node) == STARPU_OPENCL_RAM;
  778. old_is_opencl = starpu_node_get_kind(old_transfer->src_node) == STARPU_OPENCL_RAM;
  779. old_is_opencl |= starpu_node_get_kind(old_transfer->dst_node) == STARPU_OPENCL_RAM;
  780. new_is_gpu_gpu = new_transfer->src_node && new_transfer->dst_node;
  781. old_is_gpu_gpu = old_transfer->src_node && old_transfer->dst_node;
  782. /* We ignore cuda-opencl transfers, they can not happen */
  783. STARPU_ASSERT(!((new_is_cuda && old_is_opencl) || (old_is_cuda && new_is_opencl)));
  784. /* The following constraints have been observed with CUDA alone */
  785. /* Same source/destination, sequential */
  786. if (new_transfer->src_node == old_transfer->src_node && new_transfer->dst_node == old_transfer->dst_node)
  787. return 1;
  788. /* Crossed GPU-GPU, sequential */
  789. if (new_is_gpu_gpu
  790. && new_transfer->src_node == old_transfer->dst_node
  791. && old_transfer->src_node == new_transfer->dst_node)
  792. return 1;
  793. /* GPU-GPU transfers are sequential with any RAM->GPU transfer */
  794. if (new_is_gpu_gpu
  795. && (old_transfer->dst_node == new_transfer->src_node
  796. || old_transfer->dst_node == new_transfer->dst_node))
  797. return 1;
  798. if (old_is_gpu_gpu
  799. && (new_transfer->dst_node == old_transfer->src_node
  800. || new_transfer->dst_node == old_transfer->dst_node))
  801. return 1;
  802. /* StarPU's constraint on CUDA transfers is using one stream per
  803. * source/destination pair, which is already handled above */
  804. return 0;
  805. }
  806. static void transfer_queue(struct transfer *transfer)
  807. {
  808. unsigned src = transfer->src_node;
  809. unsigned dst = transfer->dst_node;
  810. struct transfer_runner *t = &transfer_runner[src][dst];
  811. if (!t->runner)
  812. {
  813. /* No runner yet, start it */
  814. static starpu_pthread_mutex_t mutex; /* process_create may yield */
  815. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  816. if (!t->runner)
  817. {
  818. char s[64];
  819. snprintf(s, sizeof(s), "transfer %u-%u runner", src, dst);
  820. starpu_pthread_create_on(s, &t->runner, NULL, transfer_execute, (void*)(uintptr_t)((src<<16) + dst), _starpu_simgrid_get_memnode_host(src));
  821. starpu_sem_init(&t->sem, 0, 0);
  822. }
  823. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  824. }
  825. if (t->last_transfer)
  826. {
  827. /* Already running a transfer, queue */
  828. t->last_transfer->next = transfer;
  829. t->last_transfer = transfer;
  830. }
  831. else
  832. {
  833. STARPU_ASSERT(!t->first_transfer);
  834. t->first_transfer = transfer;
  835. t->last_transfer = transfer;
  836. }
  837. starpu_sem_post(&t->sem);
  838. }
  839. /* Actually execute the transfer, and then start transfers waiting for this one. */
  840. static void *transfer_execute(void *arg)
  841. {
  842. unsigned src_dst = (uintptr_t) arg;
  843. unsigned src = src_dst >> 16;
  844. unsigned dst = src_dst & 0xffff;
  845. struct transfer_runner *t = &transfer_runner[src][dst];
  846. _STARPU_DEBUG("transfer runner %u-%u started\n", src, dst);
  847. while (1)
  848. {
  849. struct transfer *transfer;
  850. starpu_sem_wait(&t->sem);
  851. if (!runners_running)
  852. break;
  853. transfer = t->first_transfer;
  854. t->first_transfer = transfer->next;
  855. if (t->last_transfer == transfer)
  856. t->last_transfer = NULL;
  857. #if defined(HAVE_SG_HOST_SEND_TO) || defined(HAVE_SG_HOST_SENDTO)
  858. if (transfer->size)
  859. #else
  860. if (transfer->task)
  861. #endif
  862. {
  863. _STARPU_DEBUG("transfer %p started\n", transfer);
  864. #if defined(HAVE_SG_HOST_SEND_TO) || defined(HAVE_SG_HOST_SENDTO)
  865. #ifdef HAVE_SG_HOST_SENDTO
  866. sg_host_sendto
  867. #else
  868. sg_host_send_to
  869. #endif
  870. (_starpu_simgrid_memory_node_get_host(transfer->src_node),
  871. _starpu_simgrid_memory_node_get_host(transfer->dst_node),
  872. transfer->size);
  873. #else
  874. MSG_task_execute(transfer->task);
  875. MSG_task_destroy(transfer->task);
  876. #endif
  877. _STARPU_DEBUG("transfer %p finished\n", transfer);
  878. }
  879. *transfer->finished = 1;
  880. transfer_list_erase(&pending, transfer);
  881. /* The workers which started this request may be sleeping out of tasks, wake it */
  882. _starpu_wake_all_blocked_workers_on_node(transfer->run_node);
  883. unsigned i;
  884. /* Wake transfers waiting for my termination */
  885. /* Note: due to possible preemption inside process_create, the array
  886. * may grow while doing this */
  887. for (i = 0; i < transfer->nwake; i++)
  888. {
  889. struct transfer *wake = transfer->wake[i];
  890. STARPU_ASSERT(wake->nwait > 0);
  891. wake->nwait--;
  892. if (!wake->nwait)
  893. {
  894. _STARPU_DEBUG("triggering transfer %p\n", wake);
  895. transfer_queue(wake);
  896. }
  897. }
  898. free(transfer->wake);
  899. free(transfer);
  900. }
  901. return 0;
  902. }
  903. /* Look for sequentialization between this transfer and pending transfers, and submit this one */
  904. static void transfer_submit(struct transfer *transfer)
  905. {
  906. struct transfer *old;
  907. for (old = transfer_list_begin(&pending);
  908. old != transfer_list_end(&pending);
  909. old = transfer_list_next(old))
  910. {
  911. if (transfers_are_sequential(transfer, old))
  912. {
  913. _STARPU_DEBUG("transfer %p(%d->%d) waits for %p(%d->%d)\n",
  914. transfer, transfer->src_node, transfer->dst_node,
  915. old, old->src_node, old->dst_node);
  916. /* Make new wait for the old */
  917. transfer->nwait++;
  918. /* Make old wake the new */
  919. _STARPU_REALLOC(old->wake, (old->nwake + 1) * sizeof(old->wake));
  920. old->wake[old->nwake] = transfer;
  921. old->nwake++;
  922. }
  923. }
  924. transfer_list_push_front(&pending, transfer);
  925. if (!transfer->nwait)
  926. {
  927. _STARPU_DEBUG("transfer %p waits for nobody, starting\n", transfer);
  928. transfer_queue(transfer);
  929. }
  930. }
  931. int _starpu_simgrid_wait_transfer_event(union _starpu_async_channel_event *event)
  932. {
  933. /* this is not associated to a request so it's synchronous */
  934. starpu_pthread_wait_t wait;
  935. starpu_pthread_wait_init(&wait);
  936. starpu_pthread_queue_register(&wait, event->queue);
  937. while(1)
  938. {
  939. starpu_pthread_wait_reset(&wait);
  940. if (event->finished)
  941. break;
  942. starpu_pthread_wait_wait(&wait);
  943. }
  944. starpu_pthread_queue_unregister(&wait, event->queue);
  945. starpu_pthread_wait_destroy(&wait);
  946. return 0;
  947. }
  948. int _starpu_simgrid_test_transfer_event(union _starpu_async_channel_event *event)
  949. {
  950. return event->finished;
  951. }
  952. /* Wait for completion of all transfers */
  953. static void _starpu_simgrid_wait_transfers(void)
  954. {
  955. unsigned finished = 0;
  956. struct transfer *sync = transfer_new();
  957. struct transfer *cur;
  958. #if defined(HAVE_SG_HOST_SEND_TO) || defined(HAVE_SG_HOST_SENDTO)
  959. sync->size = 0;
  960. #else
  961. sync->task = NULL;
  962. #endif
  963. sync->finished = &finished;
  964. sync->src_node = STARPU_MAIN_RAM;
  965. sync->dst_node = STARPU_MAIN_RAM;
  966. sync->run_node = STARPU_MAIN_RAM;
  967. sync->wake = NULL;
  968. sync->nwake = 0;
  969. sync->nwait = 0;
  970. sync->next = NULL;
  971. for (cur = transfer_list_begin(&pending);
  972. cur != transfer_list_end(&pending);
  973. cur = transfer_list_next(cur))
  974. {
  975. sync->nwait++;
  976. _STARPU_REALLOC(cur->wake, (cur->nwake + 1) * sizeof(cur->wake));
  977. cur->wake[cur->nwake] = sync;
  978. cur->nwake++;
  979. }
  980. if (sync->nwait == 0)
  981. {
  982. /* No transfer to wait for */
  983. free(sync);
  984. return;
  985. }
  986. /* Push synchronization pseudo-transfer */
  987. transfer_list_push_front(&pending, sync);
  988. /* And wait for it */
  989. starpu_pthread_wait_t wait;
  990. starpu_pthread_wait_init(&wait);
  991. starpu_pthread_queue_register(&wait, &_starpu_simgrid_transfer_queue[STARPU_MAIN_RAM]);
  992. while(1)
  993. {
  994. starpu_pthread_wait_reset(&wait);
  995. if (finished)
  996. break;
  997. starpu_pthread_wait_wait(&wait);
  998. }
  999. starpu_pthread_queue_unregister(&wait, &_starpu_simgrid_transfer_queue[STARPU_MAIN_RAM]);
  1000. starpu_pthread_wait_destroy(&wait);
  1001. }
  1002. /* Data transfer issued by StarPU */
  1003. int _starpu_simgrid_transfer(size_t size, unsigned src_node, unsigned dst_node, struct _starpu_data_request *req)
  1004. {
  1005. /* Simgrid does not like 0-bytes transfers */
  1006. if (!size)
  1007. return 0;
  1008. /* Explicitly disabled by user? */
  1009. if (!simgrid_transfer_cost)
  1010. return 0;
  1011. union _starpu_async_channel_event *event, myevent;
  1012. double start = 0.;
  1013. struct transfer *transfer = transfer_new();
  1014. _STARPU_DEBUG("creating transfer %p for %lu bytes\n", transfer, (unsigned long) size);
  1015. #if defined(HAVE_SG_HOST_SEND_TO) || defined(HAVE_SG_HOST_SENDTO)
  1016. transfer->size = size;
  1017. #else
  1018. msg_task_t task;
  1019. starpu_sg_host_t *hosts;
  1020. double *computation;
  1021. double *communication;
  1022. _STARPU_CALLOC(hosts, 2, sizeof(*hosts));
  1023. _STARPU_CALLOC(computation, 2, sizeof(*computation));
  1024. _STARPU_CALLOC(communication, 4, sizeof(*communication));
  1025. hosts[0] = _starpu_simgrid_memory_node_get_host(src_node);
  1026. hosts[1] = _starpu_simgrid_memory_node_get_host(dst_node);
  1027. STARPU_ASSERT(hosts[0] != hosts[1]);
  1028. communication[1] = size;
  1029. task = MSG_parallel_task_create("copy", 2, hosts, computation, communication, NULL);
  1030. transfer->task = task;
  1031. #endif
  1032. transfer->src_node = src_node;
  1033. transfer->dst_node = dst_node;
  1034. transfer->run_node = starpu_worker_get_local_memory_node();
  1035. if (req)
  1036. event = &req->async_channel.event;
  1037. else
  1038. event = &myevent;
  1039. event->finished = 0;
  1040. transfer->finished = &event->finished;
  1041. event->queue = &_starpu_simgrid_transfer_queue[transfer->run_node];
  1042. transfer->wake = NULL;
  1043. transfer->nwake = 0;
  1044. transfer->nwait = 0;
  1045. transfer->next = NULL;
  1046. if (req)
  1047. starpu_interface_start_driver_copy_async(src_node, dst_node, &start);
  1048. /* Sleep 10µs for the GPU transfer queueing */
  1049. if (_starpu_simgrid_queue_malloc_cost())
  1050. starpu_sleep(0.000010);
  1051. transfer_submit(transfer);
  1052. /* Note: from here, transfer might be already freed */
  1053. if (req)
  1054. {
  1055. starpu_interface_end_driver_copy_async(src_node, dst_node, start);
  1056. starpu_interface_data_copy(src_node, dst_node, size);
  1057. return -EAGAIN;
  1058. }
  1059. else
  1060. {
  1061. /* this is not associated to a request so it's synchronous */
  1062. _starpu_simgrid_wait_transfer_event(event);
  1063. return 0;
  1064. }
  1065. }
  1066. /* Sync all GPUs (used on CUDA Free, typically) */
  1067. void _starpu_simgrid_sync_gpus(void)
  1068. {
  1069. _starpu_simgrid_wait_transfers();
  1070. }
  1071. int
  1072. _starpu_simgrid_thread_start(int argc STARPU_ATTRIBUTE_UNUSED, char *argv[])
  1073. {
  1074. void *(*f)(void*) = (void*) (uintptr_t) strtol(argv[0], NULL, 16);
  1075. void *arg = (void*) (uintptr_t) strtol(argv[1], NULL, 16);
  1076. /* FIXME: Ugly work-around for bug in simgrid: the MPI context is not properly set at MSG process startup */
  1077. starpu_sleep(0.000001);
  1078. _starpu_simgrid_actor_setup();
  1079. /* _args is freed with process context */
  1080. f(arg);
  1081. return 0;
  1082. }
  1083. starpu_pthread_t _starpu_simgrid_actor_create(const char *name, xbt_main_func_t code, starpu_sg_host_t host, int argc, char *argv[])
  1084. {
  1085. void **tsd;
  1086. starpu_pthread_t actor;
  1087. _STARPU_CALLOC(tsd, MAX_TSD+1, sizeof(void*));
  1088. #ifdef HAVE_SG_ACTOR_INIT
  1089. actor = sg_actor_init(name, host);
  1090. #ifdef HAVE_SG_ACTOR_SET_DATA
  1091. sg_actor_set_data(actor, tsd);
  1092. #else
  1093. sg_actor_data_set(actor, tsd);
  1094. #endif
  1095. sg_actor_start(actor, code, argc, argv);
  1096. #else
  1097. actor = MSG_process_create_with_arguments(name, code, tsd, host, argc, argv);
  1098. #ifdef HAVE_SG_ACTOR_DATA
  1099. #ifdef HAVE_SG_ACTOR_SET_DATA
  1100. sg_actor_set_data(actor, tsd);
  1101. #else
  1102. sg_actor_data_set(actor, tsd);
  1103. #endif
  1104. #endif
  1105. #endif
  1106. return actor;
  1107. }
  1108. starpu_sg_host_t _starpu_simgrid_get_memnode_host(unsigned node)
  1109. {
  1110. const char *fmt;
  1111. char name[16];
  1112. switch (starpu_node_get_kind(node))
  1113. {
  1114. case STARPU_CPU_RAM:
  1115. fmt = "RAM";
  1116. break;
  1117. case STARPU_CUDA_RAM:
  1118. fmt = "CUDA%u";
  1119. break;
  1120. case STARPU_OPENCL_RAM:
  1121. fmt = "OpenCL%u";
  1122. break;
  1123. case STARPU_DISK_RAM:
  1124. fmt = "DISK%u";
  1125. break;
  1126. default:
  1127. STARPU_ABORT();
  1128. break;
  1129. }
  1130. snprintf(name, sizeof(name), fmt, starpu_memory_node_get_devid(node));
  1131. return _starpu_simgrid_get_host_by_name(name);
  1132. }
  1133. void _starpu_simgrid_count_ngpus(void)
  1134. {
  1135. #if (defined(HAVE_SG_LINK_GET_NAME) || defined(HAVE_SG_LINK_NAME) || defined sg_link_name) && (SIMGRID_VERSION >= 31300)
  1136. unsigned src, dst;
  1137. starpu_sg_host_t ramhost = _starpu_simgrid_get_host_by_name("RAM");
  1138. /* For each pair of memory nodes, get the route */
  1139. for (src = 1; src < STARPU_MAXNODES; src++)
  1140. for (dst = 1; dst < STARPU_MAXNODES; dst++)
  1141. {
  1142. int busid;
  1143. starpu_sg_host_t srchost, dsthost;
  1144. #if defined(HAVE_SG_HOST_GET_ROUTE) || defined(HAVE_SG_HOST_ROUTE) || defined(sg_host_route)
  1145. xbt_dynar_t route_dynar = xbt_dynar_new(sizeof(SD_link_t), NULL);
  1146. SD_link_t *route;
  1147. #else
  1148. const SD_link_t *route;
  1149. #endif
  1150. int i, routesize;
  1151. int through;
  1152. unsigned src2;
  1153. unsigned ngpus;
  1154. const char *name;
  1155. if (dst == src)
  1156. continue;
  1157. busid = starpu_bus_get_id(src, dst);
  1158. if (busid == -1)
  1159. continue;
  1160. srchost = _starpu_simgrid_get_memnode_host(src);
  1161. dsthost = _starpu_simgrid_get_memnode_host(dst);
  1162. #if defined(HAVE_SG_HOST_GET_ROUTE) || defined(HAVE_SG_HOST_ROUTE) || defined(sg_host_route)
  1163. #ifdef HAVE_SG_HOST_GET_ROUTE
  1164. sg_host_get_route(srchost, dsthost, route_dynar);
  1165. #else
  1166. sg_host_route(srchost, dsthost, route_dynar);
  1167. #endif
  1168. routesize = xbt_dynar_length(route_dynar);
  1169. route = xbt_dynar_to_array(route_dynar);
  1170. #else
  1171. routesize = SD_route_get_size(srchost, dsthost);
  1172. route = SD_route_get_list(srchost, dsthost);
  1173. #endif
  1174. /* If it goes through "Host", do not care, there is no
  1175. * direct transfer support */
  1176. for (i = 0; i < routesize; i++)
  1177. if (
  1178. #ifdef HAVE_SG_LINK_GET_NAME
  1179. !strcmp(sg_link_get_name(route[i]), "Host")
  1180. #else
  1181. !strcmp(sg_link_name(route[i]), "Host")
  1182. #endif
  1183. )
  1184. break;
  1185. if (i < routesize)
  1186. continue;
  1187. /* Get the PCI bridge between down and up links */
  1188. through = -1;
  1189. for (i = 0; i < routesize; i++)
  1190. {
  1191. #ifdef HAVE_SG_LINK_GET_NAME
  1192. name = sg_link_get_name(route[i]);
  1193. #else
  1194. name = sg_link_name(route[i]);
  1195. #endif
  1196. size_t len = strlen(name);
  1197. if (!strcmp(" through", name+len-8))
  1198. through = i;
  1199. else if (!strcmp(" up", name+len-3))
  1200. break;
  1201. }
  1202. /* Didn't find it ?! */
  1203. if (through == -1)
  1204. {
  1205. _STARPU_DEBUG("Didn't find through-link for %d->%d\n", src, dst);
  1206. continue;
  1207. }
  1208. #ifdef HAVE_SG_LINK_GET_NAME
  1209. name = sg_link_get_name(route[through]);
  1210. #else
  1211. name = sg_link_name(route[through]);
  1212. #endif
  1213. /*
  1214. * count how many direct routes go through it between
  1215. * GPUs and RAM
  1216. */
  1217. ngpus = 0;
  1218. for (src2 = 1; src2 < STARPU_MAXNODES; src2++)
  1219. {
  1220. int numa;
  1221. int nnumas = starpu_memory_nodes_get_numa_count();
  1222. int found = 0;
  1223. for (numa = 0; numa < nnumas; numa++)
  1224. if (starpu_bus_get_id(src2, numa) != -1)
  1225. {
  1226. found = 1;
  1227. break;
  1228. }
  1229. if (!found)
  1230. continue;
  1231. starpu_sg_host_t srchost2 = _starpu_simgrid_get_memnode_host(src2);
  1232. int routesize2;
  1233. #if defined(HAVE_SG_HOST_GET_ROUTE) || defined(HAVE_SG_HOST_ROUTE) || defined(sg_host_route)
  1234. xbt_dynar_t route_dynar2 = xbt_dynar_new(sizeof(SD_link_t), NULL);
  1235. SD_link_t *route2;
  1236. #ifdef HAVE_SG_HOST_GET_ROUTE
  1237. sg_host_get_route(srchost2, ramhost, route_dynar2);
  1238. #else
  1239. sg_host_route(srchost2, ramhost, route_dynar2);
  1240. #endif
  1241. routesize2 = xbt_dynar_length(route_dynar2);
  1242. route2 = xbt_dynar_to_array(route_dynar2);
  1243. #else
  1244. const SD_link_t *route2 = SD_route_get_list(srchost2, ramhost);
  1245. routesize2 = SD_route_get_size(srchost2, ramhost);
  1246. #endif
  1247. for (i = 0; i < routesize2; i++)
  1248. if (
  1249. #ifdef HAVE_SG_LINK_GET_NAME
  1250. !strcmp(name, sg_link_get_name(route2[i]))
  1251. #else
  1252. !strcmp(name, sg_link_name(route2[i]))
  1253. #endif
  1254. )
  1255. {
  1256. /* This GPU goes through this PCI bridge to access RAM */
  1257. ngpus++;
  1258. break;
  1259. }
  1260. #if defined(HAVE_SG_HOST_GET_ROUTE) || defined(HAVE_SG_HOST_ROUTE) || defined(sg_host_route)
  1261. free(route2);
  1262. #endif
  1263. }
  1264. _STARPU_DEBUG("%d->%d through %s, %u GPUs\n", src, dst, name, ngpus);
  1265. starpu_bus_set_ngpus(busid, ngpus);
  1266. #if defined(HAVE_SG_HOST_GET_ROUTE) || defined(HAVE_SG_HOST_ROUTE) || defined(sg_host_route)
  1267. free(route);
  1268. #endif
  1269. }
  1270. #endif
  1271. }
  1272. #if 0
  1273. static size_t used;
  1274. void _starpu_simgrid_data_new(size_t size)
  1275. {
  1276. // Note: this is just declarative
  1277. //_STARPU_DISP("data new: %zd, now %zd\n", size, used);
  1278. }
  1279. void _starpu_simgrid_data_increase(size_t size)
  1280. {
  1281. used += size;
  1282. _STARPU_DISP("data increase: %zd, now %zd\n", size, used);
  1283. }
  1284. void _starpu_simgrid_data_alloc(size_t size)
  1285. {
  1286. used += size;
  1287. _STARPU_DISP("data alloc: %zd, now %zd\n", size, used);
  1288. }
  1289. void _starpu_simgrid_data_free(size_t size)
  1290. {
  1291. used -= size;
  1292. _STARPU_DISP("data free: %zd, now %zd\n", size, used);
  1293. }
  1294. void _starpu_simgrid_data_transfer(size_t size, unsigned src_node, unsigned dst_node)
  1295. {
  1296. _STARPU_DISP("data transfer %zd from %u to %u\n", size, src_node, dst_node);
  1297. }
  1298. #endif
  1299. void starpu_energy_use(float joules)
  1300. {
  1301. _starpu_simgrid_dynamic_energy += joules;
  1302. }
  1303. double starpu_energy_used(void)
  1304. {
  1305. float idle_power = starpu_get_env_float_default("STARPU_IDLE_POWER", 0.0);
  1306. return _starpu_simgrid_dynamic_energy + idle_power * starpu_timing_now() / 1000000;
  1307. }
  1308. #endif