simgrid.c 41 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480
  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
  345. run_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. return starpu_main(argc, argv);
  355. }
  356. static int main_ret;
  357. static _starpu_simgrid_main_ret
  358. do_starpu_main(int argc, char *argv[])
  359. {
  360. main_ret = run_starpu_main(argc, argv);
  361. _STARPU_SIMGRID_MAIN_RETURN;
  362. }
  363. /* We need it only when using smpi */
  364. #pragma weak smpi_process_get_user_data
  365. extern void *smpi_process_get_user_data();
  366. /* This is hopefully called before the application and simgrid */
  367. #undef main
  368. #pragma weak main
  369. int main(int argc, char **argv)
  370. {
  371. #ifdef HAVE_SG_CONFIG_CONTINUE_AFTER_HELP
  372. sg_config_continue_after_help();
  373. #endif
  374. if (_starpu_simgrid_running_smpi())
  375. {
  376. if (!smpi_process_get_user_data)
  377. {
  378. _STARPU_ERROR("Your version of simgrid does not provide smpi_process_get_user_data, we can not continue without it\n");
  379. }
  380. #if SIMGRID_VERSION >= 31600
  381. /* Recent versions of simgrid dlopen() us, so we don't need to
  382. * do circumvolutions, just init MPI early and run the application's main */
  383. return _starpu_mpi_simgrid_init(argc, argv);
  384. #else
  385. /* Oops, we are running old SMPI, let it start Simgrid, and we'll
  386. * take back hand in _starpu_simgrid_init from starpu_init() */
  387. return smpi_main(_starpu_mpi_simgrid_init, argc, argv);
  388. #endif
  389. }
  390. /* Already initialized? It probably has been done through a
  391. * constructor and MSG_process_attach, directly jump to real main */
  392. if (simgrid_started == 3)
  393. {
  394. return run_starpu_main(argc, argv);
  395. }
  396. /* Managed to catch application's main, initialize simgrid first */
  397. _starpu_start_simgrid(&argc, argv);
  398. simgrid_started = 2;
  399. /* Create a simgrid process for main */
  400. char **argv_cpy;
  401. _STARPU_MALLOC(argv_cpy, argc * sizeof(char*));
  402. int i;
  403. for (i = 0; i < argc; i++)
  404. argv_cpy[i] = strdup(argv[i]);
  405. /* Run the application in a separate thread */
  406. _starpu_simgrid_actor_create("main", &do_starpu_main, _starpu_simgrid_get_host_by_name("MAIN"), argc, argv_cpy);
  407. /* And run maestro in the main thread */
  408. #if defined(STARPU_SIMGRID_HAVE_SIMGRID_INIT) && defined(HAVE_SG_ACTOR_INIT)
  409. simgrid_run();
  410. #else
  411. MSG_main();
  412. #endif
  413. return main_ret;
  414. }
  415. #if defined(HAVE_MSG_PROCESS_ATTACH) || defined(MSG_process_attach) || defined(HAVE_SG_ACTOR_ATTACH)
  416. static void maestro(void *data STARPU_ATTRIBUTE_UNUSED)
  417. {
  418. #if defined(STARPU_SIMGRID_HAVE_SIMGRID_INIT) && defined(HAVE_SG_ACTOR_INIT)
  419. simgrid_run();
  420. #else
  421. MSG_main();
  422. #endif
  423. }
  424. #endif
  425. /* This is called early from starpu_init, so thread functions etc. can work */
  426. void _starpu_simgrid_init_early(int *argc STARPU_ATTRIBUTE_UNUSED, char ***argv STARPU_ATTRIBUTE_UNUSED)
  427. {
  428. #ifdef HAVE_SG_CONFIG_CONTINUE_AFTER_HELP
  429. sg_config_continue_after_help();
  430. #endif
  431. #if defined(HAVE_MSG_PROCESS_ATTACH) || defined(MSG_process_attach) || defined(HAVE_SG_ACTOR_ATTACH)
  432. if (simgrid_started < 2 && !_starpu_simgrid_running_smpi())
  433. {
  434. /* "Cannot create_maestro with this ContextFactory.
  435. * Try using --cfg=contexts/factory:thread instead."
  436. * See https://github.com/simgrid/simgrid/issues/141 */
  437. _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");
  438. #if SIMGRID_VERSION >= 31400 /* Only recent versions of simgrid support setting sg_cfg_set_string before starting simgrid */
  439. # ifdef HAVE_SG_CFG_SET_INT
  440. sg_cfg_set_string("contexts/factory", "thread");
  441. # else
  442. xbt_cfg_set_string("contexts/factory", "thread");
  443. # endif
  444. #endif
  445. /* We didn't catch application's main. */
  446. /* Start maestro as a separate thread */
  447. SIMIX_set_maestro(maestro, NULL);
  448. /* Initialize simgrid */
  449. _starpu_start_simgrid(argc, *argv);
  450. /* And attach the main thread to the main simgrid process */
  451. void **tsd;
  452. _STARPU_CALLOC(tsd, MAX_TSD+1, sizeof(void*));
  453. #if defined(HAVE_SG_ACTOR_ATTACH) && defined (HAVE_SG_ACTOR_DATA)
  454. sg_actor_t actor = sg_actor_attach("main", NULL, _starpu_simgrid_get_host_by_name("MAIN"), NULL);
  455. #ifdef HAVE_SG_ACTOR_SET_DATA
  456. sg_actor_set_data(actor, tsd);
  457. #else
  458. sg_actor_data_set(actor, tsd);
  459. #endif
  460. #else
  461. MSG_process_attach("main", tsd, _starpu_simgrid_get_host_by_name("MAIN"), NULL);
  462. #endif
  463. /* We initialized through MSG_process_attach */
  464. simgrid_started = 3;
  465. }
  466. #endif
  467. if (!simgrid_started && !starpu_main && !_starpu_simgrid_running_smpi())
  468. {
  469. /* Oops, we don't have MSG_process_attach and didn't catch the
  470. * 'main' symbol, there is no way for us */
  471. _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");
  472. }
  473. if (_starpu_simgrid_running_smpi())
  474. {
  475. #ifndef STARPU_STATIC_ONLY
  476. _STARPU_ERROR("Simgrid currently does not support privatization for dynamically-linked libraries in SMPI. Please reconfigure and build StarPU with --disable-shared");
  477. #endif
  478. #if defined(HAVE_MSG_PROCESS_USERDATA_INIT) && !defined(HAVE_SG_ACTOR_DATA)
  479. MSG_process_userdata_init();
  480. #endif
  481. void **tsd;
  482. _STARPU_CALLOC(tsd, MAX_TSD+1, sizeof(void*));
  483. #ifdef HAVE_SG_ACTOR_DATA
  484. #ifdef HAVE_SG_ACTOR_SET_DATA
  485. sg_actor_set_data(sg_actor_self(), tsd);
  486. #else
  487. sg_actor_data_set(sg_actor_self(), tsd);
  488. #endif
  489. #else
  490. smpi_process_set_user_data(tsd);
  491. #endif
  492. }
  493. unsigned i;
  494. for (i = 0; i < STARPU_MAXNODES; i++)
  495. starpu_pthread_queue_init(&_starpu_simgrid_transfer_queue[i]);
  496. for (i = 0; i < STARPU_NMAXWORKERS; i++)
  497. starpu_pthread_queue_init(&_starpu_simgrid_task_queue[i]);
  498. }
  499. /* This is called late from starpu_init, to start task executors */
  500. void _starpu_simgrid_init(void)
  501. {
  502. unsigned i;
  503. runners_running = 1;
  504. for (i = 0; i < starpu_worker_get_count(); i++)
  505. {
  506. char s[32];
  507. snprintf(s, sizeof(s), "worker %u runner", i);
  508. starpu_sem_init(&worker_runner[i].sem, 0, 0);
  509. 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)));
  510. }
  511. }
  512. void _starpu_simgrid_deinit_late(void)
  513. {
  514. #if defined(HAVE_MSG_PROCESS_ATTACH) || defined(MSG_process_attach) || defined(HAVE_SG_ACTOR_ATTACH)
  515. if (simgrid_started == 3)
  516. {
  517. /* Started with MSG_process_attach, now detach */
  518. #ifdef HAVE_SG_ACTOR_ATTACH
  519. sg_actor_detach();
  520. #else
  521. MSG_process_detach();
  522. #endif
  523. simgrid_started = 0;
  524. }
  525. #endif
  526. }
  527. void _starpu_simgrid_deinit(void)
  528. {
  529. unsigned i, j;
  530. runners_running = 0;
  531. for (i = 0; i < STARPU_MAXNODES; i++)
  532. {
  533. for (j = 0; j < STARPU_MAXNODES; j++)
  534. {
  535. struct transfer_runner *t = &transfer_runner[i][j];
  536. if (t->runner)
  537. {
  538. starpu_sem_post(&t->sem);
  539. #ifdef STARPU_HAVE_SIMGRID_ACTOR_H
  540. sg_actor_join(t->runner, 1000000);
  541. #elif SIMGRID_VERSION >= 31400
  542. MSG_process_join(t->runner, 1000000);
  543. #else
  544. starpu_sleep(1);
  545. #endif
  546. STARPU_ASSERT(t->first_transfer == NULL);
  547. STARPU_ASSERT(t->last_transfer == NULL);
  548. starpu_sem_destroy(&t->sem);
  549. }
  550. }
  551. /* FIXME: queue not empty at this point, needs proper unregistration */
  552. /* starpu_pthread_queue_destroy(&_starpu_simgrid_transfer_queue[i]); */
  553. }
  554. for (i = 0; i < starpu_worker_get_count(); i++)
  555. {
  556. struct worker_runner *w = &worker_runner[i];
  557. starpu_sem_post(&w->sem);
  558. #ifdef STARPU_HAVE_SIMGRID_ACTOR_H
  559. sg_actor_join(w->runner, 1000000);
  560. #elif SIMGRID_VERSION >= 31400
  561. MSG_process_join(w->runner, 1000000);
  562. #else
  563. starpu_sleep(1);
  564. #endif
  565. STARPU_ASSERT(w->first_task == NULL);
  566. STARPU_ASSERT(w->last_task == NULL);
  567. starpu_sem_destroy(&w->sem);
  568. starpu_pthread_queue_destroy(&_starpu_simgrid_task_queue[i]);
  569. }
  570. #if SIMGRID_VERSION >= 31300
  571. /* clean-atexit introduced in simgrid 3.13 */
  572. # ifdef HAVE_SG_CFG_SET_INT
  573. if ( sg_cfg_get_boolean("debug/clean-atexit"))
  574. # elif SIMGRID_VERSION >= 32300
  575. if ( xbt_cfg_get_boolean("debug/clean-atexit"))
  576. # else
  577. if ( xbt_cfg_get_boolean("clean-atexit"))
  578. # endif
  579. {
  580. _starpu_simgrid_deinit_late();
  581. }
  582. #endif
  583. }
  584. /*
  585. * Tasks
  586. */
  587. struct task
  588. {
  589. #ifdef HAVE_SG_ACTOR_SELF_EXECUTE
  590. double flops;
  591. #else
  592. msg_task_t task;
  593. #endif
  594. double energy;
  595. /* communication termination signalization */
  596. unsigned *finished;
  597. /* Next task on this worker */
  598. struct task *next;
  599. };
  600. /* Actually execute the task. */
  601. static void *task_execute(void *arg)
  602. {
  603. unsigned workerid = (uintptr_t) arg;
  604. struct worker_runner *w = &worker_runner[workerid];
  605. _STARPU_DEBUG("worker runner %u started\n", workerid);
  606. while (1)
  607. {
  608. struct task *task;
  609. starpu_sem_wait(&w->sem);
  610. if (!runners_running)
  611. break;
  612. task = w->first_task;
  613. w->first_task = task->next;
  614. if (w->last_task == task)
  615. w->last_task = NULL;
  616. _STARPU_DEBUG("task %p started\n", task);
  617. #ifdef HAVE_SG_ACTOR_EXECUTE
  618. sg_actor_execute(task->flops);
  619. #elif defined(HAVE_SG_ACTOR_SELF_EXECUTE)
  620. sg_actor_self_execute(task->flops);
  621. #else
  622. MSG_task_execute(task->task);
  623. MSG_task_destroy(task->task);
  624. #endif
  625. starpu_energy_use(task->energy);
  626. _STARPU_DEBUG("task %p finished\n", task);
  627. *task->finished = 1;
  628. /* The worker which started this task may be sleeping out of tasks, wake it */
  629. _starpu_wake_worker_relax(workerid);
  630. free(task);
  631. }
  632. _STARPU_DEBUG("worker %u stopped\n", workerid);
  633. return 0;
  634. }
  635. /* Wait for completion of all asynchronous tasks for this worker */
  636. void _starpu_simgrid_wait_tasks(int workerid)
  637. {
  638. struct task *task = worker_runner[workerid].last_task;
  639. if (!task)
  640. return;
  641. unsigned *finished = task->finished;
  642. starpu_pthread_wait_t wait;
  643. starpu_pthread_wait_init(&wait);
  644. starpu_pthread_queue_register(&wait, &_starpu_simgrid_task_queue[workerid]);
  645. while(1)
  646. {
  647. starpu_pthread_wait_reset(&wait);
  648. if (*finished)
  649. break;
  650. starpu_pthread_wait_wait(&wait);
  651. }
  652. starpu_pthread_queue_unregister(&wait, &_starpu_simgrid_task_queue[workerid]);
  653. starpu_pthread_wait_destroy(&wait);
  654. }
  655. /* Task execution submitted by StarPU */
  656. void _starpu_simgrid_submit_job(int workerid, int sched_ctx_id, struct _starpu_job *j, struct starpu_perfmodel_arch* perf_arch STARPU_ATTRIBUTE_UNUSED, double length, double energy, unsigned *finished)
  657. {
  658. struct starpu_task *starpu_task = j->task;
  659. double flops;
  660. #ifndef HAVE_SG_ACTOR_SELF_EXECUTE
  661. msg_task_t simgrid_task;
  662. #endif
  663. if (j->internal)
  664. /* This is not useful to include in simulation (and probably
  665. * doesn't have a perfmodel anyway) */
  666. return;
  667. if (isnan(length))
  668. {
  669. length = starpu_task_worker_expected_length(starpu_task, workerid, sched_ctx_id, j->nimpl);
  670. STARPU_ASSERT_MSG(!_STARPU_IS_ZERO(length) && !isnan(length),
  671. "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",
  672. _starpu_job_get_model_name(j), _starpu_get_perf_model_dir_codelet());
  673. /* TODO: option to add variance according to performance model,
  674. * to be able to easily check scheduling robustness */
  675. }
  676. if (isnan(energy))
  677. {
  678. energy = starpu_task_worker_expected_energy(starpu_task, workerid, sched_ctx_id, j->nimpl);
  679. /* TODO: option to add variance according to performance model,
  680. * to be able to easily check scheduling robustness */
  681. }
  682. #ifdef HAVE_SG_HOST_GET_SPEED
  683. flops = length/1000000.0*sg_host_get_speed(sg_host_self());
  684. #else
  685. #if defined(HAVE_SG_HOST_SPEED) || defined(sg_host_speed)
  686. # if defined(HAVE_SG_HOST_SELF) || defined(sg_host_self)
  687. flops = length/1000000.0*sg_host_speed(sg_host_self());
  688. # else
  689. flops = length/1000000.0*sg_host_speed(MSG_host_self());
  690. # endif
  691. #elif defined HAVE_MSG_HOST_GET_SPEED || defined(MSG_host_get_speed)
  692. flops = length/1000000.0*MSG_host_get_speed(MSG_host_self());
  693. #else
  694. flops = length/1000000.0*MSG_get_host_speed(MSG_host_self());
  695. #endif
  696. #endif
  697. #ifndef HAVE_SG_ACTOR_SELF_EXECUTE
  698. simgrid_task = MSG_task_create(_starpu_job_get_task_name(j), flops, 0, NULL);
  699. #endif
  700. if (finished == NULL)
  701. {
  702. /* Synchronous execution */
  703. /* First wait for previous tasks */
  704. _starpu_simgrid_wait_tasks(workerid);
  705. #ifdef HAVE_SG_ACTOR_EXECUTE
  706. sg_actor_execute(flops);
  707. #elif defined(HAVE_SG_ACTOR_SELF_EXECUTE)
  708. sg_actor_self_execute(flops);
  709. #else
  710. MSG_task_execute(simgrid_task);
  711. MSG_task_destroy(simgrid_task);
  712. #endif
  713. starpu_energy_use(energy);
  714. }
  715. else
  716. {
  717. /* Asynchronous execution */
  718. struct task *task;
  719. struct worker_runner *w = &worker_runner[workerid];
  720. _STARPU_MALLOC(task, sizeof(*task));
  721. #ifdef HAVE_SG_ACTOR_SELF_EXECUTE
  722. task->flops = flops;
  723. #else
  724. task->task = simgrid_task;
  725. #endif
  726. task->energy = energy;
  727. task->finished = finished;
  728. *finished = 0;
  729. task->next = NULL;
  730. /* Sleep 10µs for the GPU task queueing */
  731. if (_starpu_simgrid_queue_malloc_cost())
  732. starpu_sleep(0.000010);
  733. if (w->last_task)
  734. {
  735. /* Already running a task, queue */
  736. w->last_task->next = task;
  737. w->last_task = task;
  738. }
  739. else
  740. {
  741. STARPU_ASSERT(!w->first_task);
  742. w->first_task = task;
  743. w->last_task = task;
  744. }
  745. starpu_sem_post(&w->sem);
  746. }
  747. }
  748. /*
  749. * Transfers
  750. */
  751. /* Note: simgrid is not parallel, so there is no need to hold locks for management of transfers. */
  752. LIST_TYPE(transfer,
  753. #if defined(HAVE_SG_HOST_SEND_TO) || defined(HAVE_SG_HOST_SENDTO)
  754. size_t size;
  755. #else
  756. msg_task_t task;
  757. #endif
  758. int src_node;
  759. int dst_node;
  760. int run_node;
  761. /* communication termination signalization */
  762. unsigned *finished;
  763. /* transfers which wait for this transfer */
  764. struct transfer **wake;
  765. unsigned nwake;
  766. /* Number of transfers that this transfer waits for */
  767. unsigned nwait;
  768. /* Next transfer on this stream */
  769. struct transfer *next;
  770. )
  771. struct transfer_list pending;
  772. /* Tell for two transfers whether they should be handled in sequence */
  773. static int transfers_are_sequential(struct transfer *new_transfer, struct transfer *old_transfer)
  774. {
  775. int new_is_cuda STARPU_ATTRIBUTE_UNUSED, old_is_cuda STARPU_ATTRIBUTE_UNUSED;
  776. int new_is_opencl STARPU_ATTRIBUTE_UNUSED, old_is_opencl STARPU_ATTRIBUTE_UNUSED;
  777. int new_is_gpu_gpu, old_is_gpu_gpu;
  778. new_is_cuda = starpu_node_get_kind(new_transfer->src_node) == STARPU_CUDA_RAM;
  779. new_is_cuda |= starpu_node_get_kind(new_transfer->dst_node) == STARPU_CUDA_RAM;
  780. old_is_cuda = starpu_node_get_kind(old_transfer->src_node) == STARPU_CUDA_RAM;
  781. old_is_cuda |= starpu_node_get_kind(old_transfer->dst_node) == STARPU_CUDA_RAM;
  782. new_is_opencl = starpu_node_get_kind(new_transfer->src_node) == STARPU_OPENCL_RAM;
  783. new_is_opencl |= starpu_node_get_kind(new_transfer->dst_node) == STARPU_OPENCL_RAM;
  784. old_is_opencl = starpu_node_get_kind(old_transfer->src_node) == STARPU_OPENCL_RAM;
  785. old_is_opencl |= starpu_node_get_kind(old_transfer->dst_node) == STARPU_OPENCL_RAM;
  786. new_is_gpu_gpu = new_transfer->src_node && new_transfer->dst_node;
  787. old_is_gpu_gpu = old_transfer->src_node && old_transfer->dst_node;
  788. /* We ignore cuda-opencl transfers, they can not happen */
  789. STARPU_ASSERT(!((new_is_cuda && old_is_opencl) || (old_is_cuda && new_is_opencl)));
  790. /* The following constraints have been observed with CUDA alone */
  791. /* Same source/destination, sequential */
  792. if (new_transfer->src_node == old_transfer->src_node && new_transfer->dst_node == old_transfer->dst_node)
  793. return 1;
  794. /* Crossed GPU-GPU, sequential */
  795. if (new_is_gpu_gpu
  796. && new_transfer->src_node == old_transfer->dst_node
  797. && old_transfer->src_node == new_transfer->dst_node)
  798. return 1;
  799. /* GPU-GPU transfers are sequential with any RAM->GPU transfer */
  800. if (new_is_gpu_gpu
  801. && (old_transfer->dst_node == new_transfer->src_node
  802. || old_transfer->dst_node == new_transfer->dst_node))
  803. return 1;
  804. if (old_is_gpu_gpu
  805. && (new_transfer->dst_node == old_transfer->src_node
  806. || new_transfer->dst_node == old_transfer->dst_node))
  807. return 1;
  808. /* StarPU's constraint on CUDA transfers is using one stream per
  809. * source/destination pair, which is already handled above */
  810. return 0;
  811. }
  812. static void transfer_queue(struct transfer *transfer)
  813. {
  814. unsigned src = transfer->src_node;
  815. unsigned dst = transfer->dst_node;
  816. struct transfer_runner *t = &transfer_runner[src][dst];
  817. if (!t->runner)
  818. {
  819. /* No runner yet, start it */
  820. static starpu_pthread_mutex_t mutex; /* process_create may yield */
  821. STARPU_PTHREAD_MUTEX_LOCK(&mutex);
  822. if (!t->runner)
  823. {
  824. char s[64];
  825. snprintf(s, sizeof(s), "transfer %u-%u runner", src, dst);
  826. starpu_pthread_create_on(s, &t->runner, NULL, transfer_execute, (void*)(uintptr_t)((src<<16) + dst), _starpu_simgrid_get_memnode_host(src));
  827. starpu_sem_init(&t->sem, 0, 0);
  828. }
  829. STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
  830. }
  831. if (t->last_transfer)
  832. {
  833. /* Already running a transfer, queue */
  834. t->last_transfer->next = transfer;
  835. t->last_transfer = transfer;
  836. }
  837. else
  838. {
  839. STARPU_ASSERT(!t->first_transfer);
  840. t->first_transfer = transfer;
  841. t->last_transfer = transfer;
  842. }
  843. starpu_sem_post(&t->sem);
  844. }
  845. /* Actually execute the transfer, and then start transfers waiting for this one. */
  846. static void *transfer_execute(void *arg)
  847. {
  848. unsigned src_dst = (uintptr_t) arg;
  849. unsigned src = src_dst >> 16;
  850. unsigned dst = src_dst & 0xffff;
  851. struct transfer_runner *t = &transfer_runner[src][dst];
  852. _STARPU_DEBUG("transfer runner %u-%u started\n", src, dst);
  853. while (1)
  854. {
  855. struct transfer *transfer;
  856. starpu_sem_wait(&t->sem);
  857. if (!runners_running)
  858. break;
  859. transfer = t->first_transfer;
  860. t->first_transfer = transfer->next;
  861. if (t->last_transfer == transfer)
  862. t->last_transfer = NULL;
  863. #if defined(HAVE_SG_HOST_SEND_TO) || defined(HAVE_SG_HOST_SENDTO)
  864. if (transfer->size)
  865. #else
  866. if (transfer->task)
  867. #endif
  868. {
  869. _STARPU_DEBUG("transfer %p started\n", transfer);
  870. #if defined(HAVE_SG_HOST_SEND_TO) || defined(HAVE_SG_HOST_SENDTO)
  871. #ifdef HAVE_SG_HOST_SENDTO
  872. sg_host_sendto
  873. #else
  874. sg_host_send_to
  875. #endif
  876. (_starpu_simgrid_memory_node_get_host(transfer->src_node),
  877. _starpu_simgrid_memory_node_get_host(transfer->dst_node),
  878. transfer->size);
  879. #else
  880. MSG_task_execute(transfer->task);
  881. MSG_task_destroy(transfer->task);
  882. #endif
  883. _STARPU_DEBUG("transfer %p finished\n", transfer);
  884. }
  885. *transfer->finished = 1;
  886. transfer_list_erase(&pending, transfer);
  887. /* The workers which started this request may be sleeping out of tasks, wake it */
  888. _starpu_wake_all_blocked_workers_on_node(transfer->run_node);
  889. unsigned i;
  890. /* Wake transfers waiting for my termination */
  891. /* Note: due to possible preemption inside process_create, the array
  892. * may grow while doing this */
  893. for (i = 0; i < transfer->nwake; i++)
  894. {
  895. struct transfer *wake = transfer->wake[i];
  896. STARPU_ASSERT(wake->nwait > 0);
  897. wake->nwait--;
  898. if (!wake->nwait)
  899. {
  900. _STARPU_DEBUG("triggering transfer %p\n", wake);
  901. transfer_queue(wake);
  902. }
  903. }
  904. free(transfer->wake);
  905. free(transfer);
  906. }
  907. return 0;
  908. }
  909. /* Look for sequentialization between this transfer and pending transfers, and submit this one */
  910. static void transfer_submit(struct transfer *transfer)
  911. {
  912. struct transfer *old;
  913. for (old = transfer_list_begin(&pending);
  914. old != transfer_list_end(&pending);
  915. old = transfer_list_next(old))
  916. {
  917. if (transfers_are_sequential(transfer, old))
  918. {
  919. _STARPU_DEBUG("transfer %p(%d->%d) waits for %p(%d->%d)\n",
  920. transfer, transfer->src_node, transfer->dst_node,
  921. old, old->src_node, old->dst_node);
  922. /* Make new wait for the old */
  923. transfer->nwait++;
  924. /* Make old wake the new */
  925. _STARPU_REALLOC(old->wake, (old->nwake + 1) * sizeof(old->wake));
  926. old->wake[old->nwake] = transfer;
  927. old->nwake++;
  928. }
  929. }
  930. transfer_list_push_front(&pending, transfer);
  931. if (!transfer->nwait)
  932. {
  933. _STARPU_DEBUG("transfer %p waits for nobody, starting\n", transfer);
  934. transfer_queue(transfer);
  935. }
  936. }
  937. int _starpu_simgrid_wait_transfer_event(union _starpu_async_channel_event *event)
  938. {
  939. /* this is not associated to a request so it's synchronous */
  940. starpu_pthread_wait_t wait;
  941. starpu_pthread_wait_init(&wait);
  942. starpu_pthread_queue_register(&wait, event->queue);
  943. while(1)
  944. {
  945. starpu_pthread_wait_reset(&wait);
  946. if (event->finished)
  947. break;
  948. starpu_pthread_wait_wait(&wait);
  949. }
  950. starpu_pthread_queue_unregister(&wait, event->queue);
  951. starpu_pthread_wait_destroy(&wait);
  952. return 0;
  953. }
  954. int _starpu_simgrid_test_transfer_event(union _starpu_async_channel_event *event)
  955. {
  956. return event->finished;
  957. }
  958. /* Wait for completion of all transfers */
  959. static void _starpu_simgrid_wait_transfers(void)
  960. {
  961. unsigned finished = 0;
  962. struct transfer *sync = transfer_new();
  963. struct transfer *cur;
  964. #if defined(HAVE_SG_HOST_SEND_TO) || defined(HAVE_SG_HOST_SENDTO)
  965. sync->size = 0;
  966. #else
  967. sync->task = NULL;
  968. #endif
  969. sync->finished = &finished;
  970. sync->src_node = STARPU_MAIN_RAM;
  971. sync->dst_node = STARPU_MAIN_RAM;
  972. sync->run_node = STARPU_MAIN_RAM;
  973. sync->wake = NULL;
  974. sync->nwake = 0;
  975. sync->nwait = 0;
  976. sync->next = NULL;
  977. for (cur = transfer_list_begin(&pending);
  978. cur != transfer_list_end(&pending);
  979. cur = transfer_list_next(cur))
  980. {
  981. sync->nwait++;
  982. _STARPU_REALLOC(cur->wake, (cur->nwake + 1) * sizeof(cur->wake));
  983. cur->wake[cur->nwake] = sync;
  984. cur->nwake++;
  985. }
  986. if (sync->nwait == 0)
  987. {
  988. /* No transfer to wait for */
  989. free(sync);
  990. return;
  991. }
  992. /* Push synchronization pseudo-transfer */
  993. transfer_list_push_front(&pending, sync);
  994. /* And wait for it */
  995. starpu_pthread_wait_t wait;
  996. starpu_pthread_wait_init(&wait);
  997. starpu_pthread_queue_register(&wait, &_starpu_simgrid_transfer_queue[STARPU_MAIN_RAM]);
  998. while(1)
  999. {
  1000. starpu_pthread_wait_reset(&wait);
  1001. if (finished)
  1002. break;
  1003. starpu_pthread_wait_wait(&wait);
  1004. }
  1005. starpu_pthread_queue_unregister(&wait, &_starpu_simgrid_transfer_queue[STARPU_MAIN_RAM]);
  1006. starpu_pthread_wait_destroy(&wait);
  1007. }
  1008. /* Data transfer issued by StarPU */
  1009. int _starpu_simgrid_transfer(size_t size, unsigned src_node, unsigned dst_node, struct _starpu_data_request *req)
  1010. {
  1011. /* Simgrid does not like 0-bytes transfers */
  1012. if (!size)
  1013. return 0;
  1014. /* Explicitly disabled by user? */
  1015. if (!simgrid_transfer_cost)
  1016. return 0;
  1017. union _starpu_async_channel_event *event, myevent;
  1018. double start = 0.;
  1019. struct transfer *transfer = transfer_new();
  1020. _STARPU_DEBUG("creating transfer %p for %lu bytes\n", transfer, (unsigned long) size);
  1021. #if defined(HAVE_SG_HOST_SEND_TO) || defined(HAVE_SG_HOST_SENDTO)
  1022. transfer->size = size;
  1023. #else
  1024. msg_task_t task;
  1025. starpu_sg_host_t *hosts;
  1026. double *computation;
  1027. double *communication;
  1028. _STARPU_CALLOC(hosts, 2, sizeof(*hosts));
  1029. _STARPU_CALLOC(computation, 2, sizeof(*computation));
  1030. _STARPU_CALLOC(communication, 4, sizeof(*communication));
  1031. hosts[0] = _starpu_simgrid_memory_node_get_host(src_node);
  1032. hosts[1] = _starpu_simgrid_memory_node_get_host(dst_node);
  1033. STARPU_ASSERT(hosts[0] != hosts[1]);
  1034. communication[1] = size;
  1035. task = MSG_parallel_task_create("copy", 2, hosts, computation, communication, NULL);
  1036. transfer->task = task;
  1037. #endif
  1038. transfer->src_node = src_node;
  1039. transfer->dst_node = dst_node;
  1040. transfer->run_node = starpu_worker_get_local_memory_node();
  1041. if (req)
  1042. event = &req->async_channel.event;
  1043. else
  1044. event = &myevent;
  1045. event->finished = 0;
  1046. transfer->finished = &event->finished;
  1047. event->queue = &_starpu_simgrid_transfer_queue[transfer->run_node];
  1048. transfer->wake = NULL;
  1049. transfer->nwake = 0;
  1050. transfer->nwait = 0;
  1051. transfer->next = NULL;
  1052. if (req)
  1053. starpu_interface_start_driver_copy_async(src_node, dst_node, &start);
  1054. /* Sleep 10µs for the GPU transfer queueing */
  1055. if (_starpu_simgrid_queue_malloc_cost())
  1056. starpu_sleep(0.000010);
  1057. transfer_submit(transfer);
  1058. /* Note: from here, transfer might be already freed */
  1059. if (req)
  1060. {
  1061. starpu_interface_end_driver_copy_async(src_node, dst_node, start);
  1062. starpu_interface_data_copy(src_node, dst_node, size);
  1063. return -EAGAIN;
  1064. }
  1065. else
  1066. {
  1067. /* this is not associated to a request so it's synchronous */
  1068. _starpu_simgrid_wait_transfer_event(event);
  1069. return 0;
  1070. }
  1071. }
  1072. /* Sync all GPUs (used on CUDA Free, typically) */
  1073. void _starpu_simgrid_sync_gpus(void)
  1074. {
  1075. _starpu_simgrid_wait_transfers();
  1076. }
  1077. _starpu_simgrid_main_ret
  1078. _starpu_simgrid_thread_start(int argc STARPU_ATTRIBUTE_UNUSED, char *argv[])
  1079. {
  1080. void *(*f)(void*) = (void*) (uintptr_t) strtol(argv[0], NULL, 16);
  1081. void *arg = (void*) (uintptr_t) strtol(argv[1], NULL, 16);
  1082. /* FIXME: Ugly work-around for bug in simgrid: the MPI context is not properly set at MSG process startup */
  1083. starpu_sleep(0.000001);
  1084. _starpu_simgrid_actor_setup();
  1085. /* _args is freed with process context */
  1086. f(arg);
  1087. _STARPU_SIMGRID_MAIN_RETURN;
  1088. }
  1089. starpu_pthread_t _starpu_simgrid_actor_create(const char *name, xbt_main_func_t code, starpu_sg_host_t host, int argc, char *argv[])
  1090. {
  1091. void **tsd;
  1092. starpu_pthread_t actor;
  1093. _STARPU_CALLOC(tsd, MAX_TSD+1, sizeof(void*));
  1094. #ifdef HAVE_SG_ACTOR_INIT
  1095. actor = sg_actor_init(name, host);
  1096. #ifdef HAVE_SG_ACTOR_SET_DATA
  1097. sg_actor_set_data(actor, tsd);
  1098. #else
  1099. sg_actor_data_set(actor, tsd);
  1100. #endif
  1101. sg_actor_start(actor, code, argc, argv);
  1102. #else
  1103. actor = MSG_process_create_with_arguments(name, code, tsd, host, argc, argv);
  1104. #ifdef HAVE_SG_ACTOR_DATA
  1105. #ifdef HAVE_SG_ACTOR_SET_DATA
  1106. sg_actor_set_data(actor, tsd);
  1107. #else
  1108. sg_actor_data_set(actor, tsd);
  1109. #endif
  1110. #endif
  1111. #endif
  1112. return actor;
  1113. }
  1114. starpu_sg_host_t _starpu_simgrid_get_memnode_host(unsigned node)
  1115. {
  1116. const char *fmt;
  1117. char name[16];
  1118. switch (starpu_node_get_kind(node))
  1119. {
  1120. case STARPU_CPU_RAM:
  1121. fmt = "RAM";
  1122. break;
  1123. case STARPU_CUDA_RAM:
  1124. fmt = "CUDA%u";
  1125. break;
  1126. case STARPU_OPENCL_RAM:
  1127. fmt = "OpenCL%u";
  1128. break;
  1129. case STARPU_DISK_RAM:
  1130. fmt = "DISK%u";
  1131. break;
  1132. default:
  1133. STARPU_ABORT();
  1134. break;
  1135. }
  1136. snprintf(name, sizeof(name), fmt, starpu_memory_node_get_devid(node));
  1137. return _starpu_simgrid_get_host_by_name(name);
  1138. }
  1139. void _starpu_simgrid_count_ngpus(void)
  1140. {
  1141. #if (defined(HAVE_SG_LINK_GET_NAME) || defined(HAVE_SG_LINK_NAME) || defined sg_link_name) && (SIMGRID_VERSION >= 31300)
  1142. unsigned src, dst;
  1143. starpu_sg_host_t ramhost = _starpu_simgrid_get_host_by_name("RAM");
  1144. /* For each pair of memory nodes, get the route */
  1145. for (src = 1; src < STARPU_MAXNODES; src++)
  1146. for (dst = 1; dst < STARPU_MAXNODES; dst++)
  1147. {
  1148. int busid;
  1149. starpu_sg_host_t srchost, dsthost;
  1150. xbt_dynar_t route_dynar = xbt_dynar_new(sizeof(SD_link_t), NULL);
  1151. SD_link_t link;
  1152. int i, routesize;
  1153. int through;
  1154. unsigned src2;
  1155. unsigned ngpus;
  1156. const char *name;
  1157. if (dst == src)
  1158. continue;
  1159. busid = starpu_bus_get_id(src, dst);
  1160. if (busid == -1)
  1161. continue;
  1162. srchost = _starpu_simgrid_get_memnode_host(src);
  1163. dsthost = _starpu_simgrid_get_memnode_host(dst);
  1164. #if defined(HAVE_SG_HOST_GET_ROUTE) || defined(HAVE_SG_HOST_ROUTE) || defined(sg_host_route)
  1165. #ifdef HAVE_SG_HOST_GET_ROUTE
  1166. sg_host_get_route(srchost, dsthost, route_dynar);
  1167. #else
  1168. sg_host_route(srchost, dsthost, route_dynar);
  1169. #endif
  1170. routesize = xbt_dynar_length(route_dynar);
  1171. #else
  1172. const SD_link_t *route = SD_route_get_list(srchost, dsthost);
  1173. routesize = SD_route_get_size(srchost, dsthost);
  1174. for (i = 0; i < routesize; i++)
  1175. xbt_dynar_push(route_dynar, &route[i]);
  1176. free(route);
  1177. #endif
  1178. /* If it goes through "Host", do not care, there is no
  1179. * direct transfer support */
  1180. for (i = 0; i < routesize; i++)
  1181. {
  1182. xbt_dynar_get_cpy(route_dynar, i, &link);
  1183. if (
  1184. #ifdef HAVE_SG_LINK_GET_NAME
  1185. !strcmp(sg_link_get_name(link), "Host")
  1186. #else
  1187. !strcmp(sg_link_name(link), "Host")
  1188. #endif
  1189. )
  1190. break;
  1191. }
  1192. if (i < routesize)
  1193. continue;
  1194. /* Get the PCI bridge between down and up links */
  1195. through = -1;
  1196. for (i = 0; i < routesize; i++)
  1197. {
  1198. xbt_dynar_get_cpy(route_dynar, i, &link);
  1199. #ifdef HAVE_SG_LINK_GET_NAME
  1200. name = sg_link_get_name(link);
  1201. #else
  1202. name = sg_link_name(link);
  1203. #endif
  1204. size_t len = strlen(name);
  1205. if (!strcmp(" through", name+len-8))
  1206. through = i;
  1207. else if (!strcmp(" up", name+len-3))
  1208. break;
  1209. }
  1210. /* Didn't find it ?! */
  1211. if (through == -1)
  1212. {
  1213. _STARPU_DEBUG("Didn't find through-link for %d->%d\n", src, dst);
  1214. continue;
  1215. }
  1216. xbt_dynar_get_cpy(route_dynar, through, &link);
  1217. #ifdef HAVE_SG_LINK_GET_NAME
  1218. name = sg_link_get_name(link);
  1219. #else
  1220. name = sg_link_name(link);
  1221. #endif
  1222. /*
  1223. * count how many direct routes go through it between
  1224. * GPUs and RAM
  1225. */
  1226. ngpus = 0;
  1227. for (src2 = 1; src2 < STARPU_MAXNODES; src2++)
  1228. {
  1229. int numa;
  1230. int nnumas = starpu_memory_nodes_get_numa_count();
  1231. int found = 0;
  1232. for (numa = 0; numa < nnumas; numa++)
  1233. if (starpu_bus_get_id(src2, numa) != -1)
  1234. {
  1235. found = 1;
  1236. break;
  1237. }
  1238. if (!found)
  1239. continue;
  1240. starpu_sg_host_t srchost2 = _starpu_simgrid_get_memnode_host(src2);
  1241. int routesize2;
  1242. xbt_dynar_t route_dynar2 = xbt_dynar_new(sizeof(SD_link_t), NULL);
  1243. #if defined(HAVE_SG_HOST_GET_ROUTE) || defined(HAVE_SG_HOST_ROUTE) || defined(sg_host_route)
  1244. #ifdef HAVE_SG_HOST_GET_ROUTE
  1245. sg_host_get_route(srchost2, ramhost, route_dynar2);
  1246. #else
  1247. sg_host_route(srchost2, ramhost, route_dynar2);
  1248. #endif
  1249. routesize2 = xbt_dynar_length(route_dynar2);
  1250. #else
  1251. const SD_link_t *route2 = SD_route_get_list(srchost2, ramhost);
  1252. routesize2 = SD_route_get_size(srchost2, ramhost);
  1253. for (i = 0; i < routesize2; i++)
  1254. xbt_dynar_push(route_dynar2, &route2[i]);
  1255. free(route2);
  1256. #endif
  1257. for (i = 0; i < routesize2; i++)
  1258. {
  1259. xbt_dynar_get_cpy(route_dynar, i, &link);
  1260. if (
  1261. #ifdef HAVE_SG_LINK_GET_NAME
  1262. !strcmp(name, sg_link_get_name(link))
  1263. #else
  1264. !strcmp(name, sg_link_name(link))
  1265. #endif
  1266. )
  1267. {
  1268. /* This GPU goes through this PCI bridge to access RAM */
  1269. ngpus++;
  1270. break;
  1271. }
  1272. }
  1273. }
  1274. _STARPU_DEBUG("%d->%d through %s, %u GPUs\n", src, dst, name, ngpus);
  1275. starpu_bus_set_ngpus(busid, ngpus);
  1276. }
  1277. #endif
  1278. }
  1279. #if 0
  1280. static size_t used;
  1281. void _starpu_simgrid_data_new(size_t size)
  1282. {
  1283. // Note: this is just declarative
  1284. //_STARPU_DISP("data new: %zd, now %zd\n", size, used);
  1285. }
  1286. void _starpu_simgrid_data_increase(size_t size)
  1287. {
  1288. used += size;
  1289. _STARPU_DISP("data increase: %zd, now %zd\n", size, used);
  1290. }
  1291. void _starpu_simgrid_data_alloc(size_t size)
  1292. {
  1293. used += size;
  1294. _STARPU_DISP("data alloc: %zd, now %zd\n", size, used);
  1295. }
  1296. void _starpu_simgrid_data_free(size_t size)
  1297. {
  1298. used -= size;
  1299. _STARPU_DISP("data free: %zd, now %zd\n", size, used);
  1300. }
  1301. void _starpu_simgrid_data_transfer(size_t size, unsigned src_node, unsigned dst_node)
  1302. {
  1303. _STARPU_DISP("data transfer %zd from %u to %u\n", size, src_node, dst_node);
  1304. }
  1305. #endif
  1306. void starpu_energy_use(float joules)
  1307. {
  1308. _starpu_simgrid_dynamic_energy += joules;
  1309. }
  1310. double starpu_energy_used(void)
  1311. {
  1312. float idle_power = starpu_get_env_float_default("STARPU_IDLE_POWER", 0.0);
  1313. return _starpu_simgrid_dynamic_energy + idle_power * starpu_timing_now() / 1000000;
  1314. }
  1315. #endif