Arcane  v4.1.10.0
Documentation développeur
Chargement...
Recherche...
Aucune correspondance
ArcaneMainBatch.cc
1// -*- tab-width: 2; indent-tabs-mode: nil; coding: utf-8-with-signature -*-
2//-----------------------------------------------------------------------------
3// Copyright 2000-2026 CEA (www.cea.fr) IFPEN (www.ifpenergiesnouvelles.com)
4// See the top-level COPYRIGHT file for details.
5// SPDX-License-Identifier: Apache-2.0
6//-----------------------------------------------------------------------------
7/*---------------------------------------------------------------------------*/
8/* ArcaneMainBatch.cc (C) 2000-2026 */
9/* */
10/* Gestion de l'exécution en mode Batch. */
11/*---------------------------------------------------------------------------*/
12/*---------------------------------------------------------------------------*/
13
14#include "arcane/utils/Iostream.h"
15#include "arcane/utils/Ptr.h"
16#include "arcane/utils/StdHeader.h"
17#include "arcane/utils/PlatformUtils.h"
18#include "arcane/utils/List.h"
19#include "arcane/utils/ApplicationInfo.h"
20#include "arcane/utils/NotSupportedException.h"
21#include "arcane/utils/FatalErrorException.h"
22#include "arcane/utils/String.h"
23#include "arcane/utils/ITraceMng.h"
24#include "arcane/utils/OStringStream.h"
25#include "arcane/utils/IMemoryInfo.h"
26#include "arcane/utils/Array.h"
27#include "arcane/utils/IFunctor.h"
28#include "arcane/utils/StringBuilder.h"
29#include "arcane/utils/ScopedPtr.h"
30#include "arcane/utils/ValueConvert.h"
31#include "arcane/utils/IProcessorAffinityService.h"
32#include "arcane/utils/ArgumentException.h"
33#include "arcane/utils/CStringUtils.h"
34#include "arcane/utils/ITraceMngPolicy.h"
35#include "arcane/utils/CommandLineArguments.h"
36#include "arcane/utils/CriticalSection.h"
37#include "arccore/common/internal/ParameterListPropertyReader.h"
38#include "arccore/common/internal/Property.h"
39
40#include "arcane/impl/ArcaneMain.h"
41#include "arcane/impl/ParallelReplication.h"
42
43#include "arcane/core/IIOMng.h"
44#include "arcane/core/ICodeService.h"
45#include "arcane/core/ISession.h"
46#include "arcane/core/Timer.h"
47#include "arcane/core/ISubDomain.h"
48#include "arcane/core/IApplication.h"
49#include "arcane/core/ITimeLoopMng.h"
50#include "arcane/core/ITimeStats.h"
51#include "arcane/core/SequentialSection.h"
52#include "arcane/core/IParallelSuperMng.h"
53#include "arcane/core/ITimeHistoryMng.h"
54#include "arcane/core/IDirectExecution.h"
55#include "arcane/core/IDirectSubDomainExecuteFunctor.h"
56#include "arcane/core/ICaseMng.h"
57#include "arcane/core/ServiceFinder2.h"
58#include "arcane/core/SubDomainBuildInfo.h"
59#include "arcane/core/IParallelMng.h"
60#include "arcane/core/IMainFactory.h"
61#include "arcane/core/ApplicationBuildInfo.h"
62#include "arcane/core/CaseDatasetSource.h"
63
64#include "arcane/core/ServiceUtils.h"
65
66#include "arcane/core/IVariableMng.h"
67#include "arcane/core/VariableCollection.h"
68#include "arcane/core/internal/IVariableMngInternal.h"
69
70#include "arcane/impl/ExecutionStatsDumper.h"
71#include "arcane/impl/TimeLoopReader.h"
72
73#include "arccore/common/accelerator/internal/RunnerInternal.h"
74
75#include <thread>
76
77/*---------------------------------------------------------------------------*/
78/*---------------------------------------------------------------------------*/
79
80namespace Arcane
81{
82
83/*---------------------------------------------------------------------------*/
84/*---------------------------------------------------------------------------*/
85
88{
89 ARCANE_DECLARE_PROPERTY_CLASS(ArcaneMainBatchProperties);
90 public:
91 Int32 m_max_iteration = 0;
92 bool m_is_continue = false;
94};
95
96/*---------------------------------------------------------------------------*/
97/*---------------------------------------------------------------------------*/
101class ArcaneMainBatch
102: public ArcaneMain
103{
104 public:
105
107 class SessionExec
108 {
110 class SubInfo
111 {
112 public:
113 SubInfo()
114 : m_sub_domain(nullptr), m_time_stats(nullptr), m_want_print_stats(false) {}
115 ~SubInfo()
116 {
117 // Il faut d'abord détruire la ITimeStats car il utilise
118 // les TimerMng des IParallelMng.
119 delete m_time_stats;
120 //delete m_rank_parallel_mng;
121 //m_world_parallel_mng.reset();
122 // Le sous-domaine est détruit lorsque la session se termine
123 }
124 SubInfo(const SubInfo&) = delete;
125 void operator=(const SubInfo&) = delete;
126 public:
127 Ref<IParallelMng> m_world_parallel_mng;
128 Ref<IParallelMng> m_rank_parallel_mng;
129 ISubDomain* m_sub_domain;
130 ITimeStats* m_time_stats;
131 bool m_want_print_stats;
132 };
133
134 public:
135 SessionExec(ArcaneMainBatch* arcane_main,ISession* session,Int32 nb_local_rank)
136 : m_arcane_main(arcane_main),
137 m_session(session),
139 m_direct_test_name(m_arcane_main->m_direct_test_name),
140 m_properties(m_arcane_main->m_properties),
141 m_code_service(m_arcane_main->m_code_service),
142 m_sub_infos(nb_local_rank),
143 m_direct_sub_domain_execute_functor(m_arcane_main->_directExecuteFunctor())
144 {
145 const CaseDatasetSource& dataset_source = m_arcane_main->applicationBuildInfo().caseDatasetSource();
146 m_case_file = dataset_source.fileName();
147 m_case_bytes = dataset_source.content();
148 // Les sub_infos de chaque thread sont créés executeRank()
149 m_sub_infos.fill(nullptr);
150 }
152 {
153 for( Integer i=0, n=m_sub_infos.size(); i<n; ++i )
154 delete m_sub_infos[i];
155 }
156 public:
157 // Collective sur les threads du processus
158 void executeRank(Int32 local_rank);
159 private:
160 IApplication* _application() { return m_arcane_main->application(); }
161 private:
162 ArcaneMainBatch* m_arcane_main;
163 ISession* m_session;
165 String m_direct_test_name;
170 UniqueArray<SubInfo*> m_sub_infos;
171 IDirectSubDomainExecuteFunctor* m_direct_sub_domain_execute_functor;
172 private:
173 void _execDirectTest(IParallelMng* pm,const String& test_name,bool is_collective);
174 void _printStats(ISubDomain* sd,ITraceMng* trace,ITimeStats* time_stat);
175 void _createAndRunSubDomain(SubInfo* sub_info,Ref<IParallelMng> pm,Ref<IParallelMng> all_replica_pm,Int32 local_rank);
176 };
177
178 class ExecFunctor
179 : public IFunctor
180 {
181 public:
182 ExecFunctor(SessionExec* session_exec,Int32 local_rank)
183 : m_session_exec(session_exec), m_local_rank(local_rank)
184 {
185 }
186 public:
187 void executeFunctor() override
188 {
189 m_session_exec->executeRank(m_local_rank);
190 }
191 private:
192 SessionExec* m_session_exec;
193 Int32 m_local_rank;
194 };
195
196 public:
197
199 ~ArcaneMainBatch() override;
200
201 void build() override;
202 void initialize() override;
203 int execute() override;
204 void doAbort() override;
205 bool parseArgs(StringList args) override;
206 void finalize() override;
207
208 private:
209
210 ISession* m_session = nullptr;
216 String m_direct_exec_name;
217 String m_direct_test_name;
219 SessionExec* m_session_exec = nullptr;
220
221 private:
222
223 bool _sequentialParseArgs(StringList args);
224};
225
226/*---------------------------------------------------------------------------*/
227/*---------------------------------------------------------------------------*/
228
229extern "C++" ARCANE_IMPL_EXPORT IArcaneMain*
230createArcaneMainBatch(const ApplicationInfo& app_info,IMainFactory* main_factory)
231{
232 return new ArcaneMainBatch(app_info,main_factory);
233}
234
235/*---------------------------------------------------------------------------*/
236/*---------------------------------------------------------------------------*/
237
238ArcaneMainBatch::
239ArcaneMainBatch(const ApplicationInfo& exe_info,IMainFactory* main_factory)
240: ArcaneMain(exe_info,main_factory)
241, m_init_only(false)
242, m_check_case_only(false)
243, m_has_sub_domain_threads(false)
244{
245}
246
247/*---------------------------------------------------------------------------*/
248/*---------------------------------------------------------------------------*/
249
251build()
252{
254}
255
256/*---------------------------------------------------------------------------*/
257/*---------------------------------------------------------------------------*/
258
264
265/*---------------------------------------------------------------------------*/
266/*---------------------------------------------------------------------------*/
267
268ArcaneMainBatch::
269~ArcaneMainBatch()
270{
271 // Normalement finalize() doit avoir été appelé pour libérer les
272 // différents objets (m_session, m_code_service, ...).
273 // Si ce n'est pas le cas, c'est probablement du à une exception et dans
274 // ce cas on ne fait rien pour éviter de détruire des objets dont on ne
275 // connait pas trop l'état interne.
276}
277
278/*---------------------------------------------------------------------------*/
279/*---------------------------------------------------------------------------*/
280
283{
284 if (ArcaneMain::parseArgs(args))
285 return true;
286
287 bool r = _sequentialParseArgs(args);
288 return r;
289}
290
291
292/*****************************************************************************
293 * Les variables ARCANE_NB_SUB_DOMAIN & ARCANE_IDLE_SERVICE sont moins prioritaires
294 * que les arguments passés à l'exécutable.
295 *****************************************************************************/
296bool ArcaneMainBatch::
297_sequentialParseArgs(StringList args)
298{
299 ITraceMng* trace = _application()->traceMng();
300
301 String us_arcane_opt("-arcane_opt");
302 String us_init_only("init_only");
303 String us_check_case_only("check_case_only");
304 String us_continue("continue");
305 String us_max_iteration("max_iteration");
306 String us_casename("casename");
307 String us_direct_exec("direct_exec");
308 String us_direct_test("direct_test");
309 String us_direct_mesh("direct_mesh");
310 String us_tool_arg("tool_arg");
311 String us_direct_exec_mesh_arg("direct_exec_mesh_arg");
312 String us_nb_sub_domain("nb_sub_domain");
313 String us_nb_replication("nb_replication");
314 String us_idle_service("idle_service");
315
316 // Remplit 'm_properties' en fonction des paramètres de la ligne de commande
317 // TODO: Ce mécanisme est disponible depuis janvier 2021. A terme, il faudra
318 // rendre obsolète et supprimer la possibilité de spécifier les options
319 // via '-arcane_opt'.
320 properties::readFromParameterList(applicationInfo().commandLineArguments().parameters(),m_properties);
321
322 CaseDatasetSource& dataset_source = _applicationBuildInfo().caseDatasetSource();
323 // Indique si on a un jeu de données.
324 bool has_case_dataset_content = !(dataset_source.fileName().empty() && dataset_source.content().empty());
325 Integer nb_arg = args.count();
326 if (nb_arg<2 && !has_case_dataset_content){
327 trace->info() << "Usage: programm input_data ; for more information: program -arcane_opt help";
328 trace->pfatal() << "No input data specified.";
329 }
330
331 StringList unknown_args;
332 StringBuilder tool_args_xml;
333 StringBuilder direct_exec_mesh_args_xml;
334 String tool_mesh;
335
336 String nb_sub_domain_str;
337 String nb_replication_str;
338 String idle_service_name = platform::getEnvironmentVariable("ARCANE_IDLE_SERVICE");
339 if (!idle_service_name.null())
340 m_properties.m_idle_service_name = idle_service_name;
341
342 for( Integer i=1, s=nb_arg-1; i<s; ++i ){
343 // cerr << "** ARGS ARGS " << i << ' ' << args[i] << '\n';
344 if (args[i]!=us_arcane_opt){
345 unknown_args.add(args[i]);
346 continue;
347 }
348 bool is_valid_opt = false;
349 ++i;
350 String str;
351 if (i<s)
352 str = args[i];
353 if (str==us_init_only){
354 m_init_only = true;
355 is_valid_opt = true;
356 }
357 else if (str==us_check_case_only){
358 m_check_case_only = true;
359 is_valid_opt = true;
360 }
361 else if (str==us_continue){
362 m_properties.m_is_continue = true;
363 is_valid_opt = true;
364 }
365 else if (str==us_max_iteration){
366 ++i;
367 if (i<s){
368 m_properties.m_max_iteration = CStringUtils::toInteger(args[i].localstr());
369 //cerr << "** MAX ITER " << m_max_iteration << '\n';
370 is_valid_opt = true;
371 }
372 else
373 trace->pfatal() << "Option 'max_iteration' must specify the number of iterations";
374 }
375 // Nom du cas.
376 else if (str==us_casename){
377 ++i;
378 if (i<s){
379 m_case_name = args[i];
380 is_valid_opt = true;
381 }
382 }
383 else if (str==us_direct_exec){
384 ++i;
385 if (i<s){
386 m_direct_exec_name = args[i];
387 //trace->info()<<"[ArcaneMainBatch] m_direct_exec_name="<<args[i];
388 is_valid_opt = true;
389 }
390 }
391 else if (str==us_direct_test){
392 ++i;
393 if (i<s){
394 m_direct_test_name = args[i];
395 //trace->info()<<"[ArcaneMainBatch] m_direct_test_name="<<args[i];
396 is_valid_opt = true;
397 }
398 }
399 else if (str==us_tool_arg || str==us_direct_exec_mesh_arg){
400 ++i;
401 String arg;
402 String value;
403 if (i<s){
404 arg = args[i];
405 }
406 ++i;
407 if (i<s){
408 value = args[i];
409 is_valid_opt = true;
410 String to_add = String::format("<{0}>{1}</{2}>\n",arg,value,arg);
411 if (str==us_tool_arg)
412 tool_args_xml += to_add;
413 else if (str==us_direct_exec_mesh_arg)
414 direct_exec_mesh_args_xml += to_add;
415 }
416 }
417 else if (str==us_nb_sub_domain){
418 ++i;
419 if (i<s){
420 nb_sub_domain_str = args[i];
421 //trace->info()<<"[ArcaneMainBatch] nb_sub_domain_str="<<args[i];
422 is_valid_opt = true;
423 }
424 }
425 else if (str==us_nb_replication){
426 ++i;
427 if (i<s){
428 nb_replication_str = args[i];
429 //trace->info()<<"[ArcaneMainBatch] nb_sub_domain_str="<<args[i];
430 is_valid_opt = true;
431 }
432 }
433 else if (str==us_idle_service){
434 ++i;
435 if (i<s){
436 m_properties.m_idle_service_name = args[i];
437 //trace->info()<<"[ArcaneMainBatch] m_idle_service_name="<<args[i];
438 is_valid_opt = true;
439 }
440 }
441 if (!is_valid_opt){
442 trace->pfatal() << "Unknown Arcane option <" << str << ">\n";
443 }
444 }
445
446 bool use_direct_test = (!m_direct_test_name.null());
447 bool use_direct_exec = (!m_direct_exec_name.null());
448
449 if (use_direct_test){
450 }
451 else if (use_direct_exec){
452 // Dans ce cas, le dernier argument de la ligne de commande est
453 // le nom du maillage.
454 tool_mesh = args[nb_arg-1];
455 dataset_source.setFileName("Dummy.arc");
456 }
457 else{
458 // Le nom du cas est contenu dans le dernier argument de la ligne
459 // de commande. On prend cet argument sauf si un nom de fichier
460 // a déjà été positionné avant d'initialiser Arcane.
461 if (dataset_source.fileName().empty() && dataset_source.content().empty())
462 dataset_source.setFileName(args[nb_arg-1]);
463 }
464
465 if (!nb_sub_domain_str.null()){
466 Int32 nb_sub_domain = 0;
467 bool is_bad = builtInGetValue(nb_sub_domain,nb_sub_domain_str);
468 if (is_bad || nb_sub_domain<=0){
469 trace->pfatal() << "Invalid number of subdomains : " << nb_sub_domain;
470 }
471 trace->info() << "Use '" << nb_sub_domain << "' subdomains";
472 _applicationBuildInfo().setNbProcessusSubDomain(nb_sub_domain);
473 }
474
475 if (!nb_replication_str.null()){
476 Int32 nb_replication = 0;
477 bool is_bad = builtInGetValue(nb_replication,nb_replication_str);
478 if (is_bad || nb_replication<0){
479 trace->pfatal() << "Invalid number of replication : " << nb_replication;
480 }
481 trace->info() << "Use replication of subdomains nb_replication=" << nb_replication;
482 _applicationBuildInfo().setNbReplicationSubDomain(nb_replication);
483 }
484
485 if (_applicationBuildInfo().nbReplicationSubDomain()!=0 && _applicationBuildInfo().nbProcessusSubDomain()!=0)
486 trace->pfatal() << "The subdomains number of replication and restriction options are incompatible.";
487
488 if (!use_direct_test){
489 String case_file = dataset_source.fileName();
490 //trace->info()<<"[ArcaneMainBatch] !use_direct_test, getCodeService";
491 m_code_service = _application()->getCodeService(case_file);
492
493 if (!m_code_service){
494 trace->info() << "The file `" << case_file << "' is not a known file type.";
495 case_file = args[nb_arg-2];
496
497 m_code_service = _application()->getCodeService(case_file);
498 if (!m_code_service){
499 trace->pfatal() << "File extension not valid.";
500 }
501 }
502 }
503
504 if (use_direct_exec){
505 //trace->info()<<"[ArcaneMainBatch] use_direct_test!";
506 // Analyse les arguments qui correspondent aux options d'exécution directes
507 // et construit un fichier xml à partir de la.
508 StringBuilder s;
509 s += "<?xml version=\"1.0\"?>\n";
510 s += "<case codename=\"ArcaneDriver\" xml:lang=\"en\" codeversion=\"1.0\">";
511 s += " <arcane>\n";
512 s += " <title>DirectExec</title>\n";
513 s += " <description>DirectExec</description>\n";
514 s += " <timeloop>ArcaneDirectExecutionLoop</timeloop>\n";
515 s += " </arcane>\n";
516 s += " <meshes>\n";
517 s += " <mesh>\n";
518 s += String::format(" <filename>{0}</filename>\n",tool_mesh);
519 s += direct_exec_mesh_args_xml;
520 s += " </mesh>\n";
521 s += " </meshes>\n";
522 s += " <arcane-direct-execution>\n";
523 s += String::format(" <tool name='{0}'>\n",m_direct_exec_name);
524 s += tool_args_xml;
525 s += " </tool>\n";
526 s += " </arcane-direct-execution>\n";
527 s += "</case>\n";
528 dataset_source.setFileName("(None)");
529 String buf = s;
530 dataset_source.setContent(buf.utf8());
531 trace->info() << "Direct exec xml file=" << s;
532 }
533
534 if (m_code_service.get()){
535 bool is_bad = m_code_service->parseArgs(unknown_args);
536 if (is_bad)
537 return true;
538 }
539
540 if (!unknown_args.empty()){
541 trace->info()<< "Unknown command line option: " << unknown_args[0];
542 }
543
544 return false;
545}
546
547/*---------------------------------------------------------------------------*/
548/*---------------------------------------------------------------------------*/
549
550namespace
551{
552struct LaunchThreadInfo
553{
554 ArcaneMainBatch* arcane_main;
555 ArcaneMainBatch::SessionExec* session_exec;
556 IApplication* application;
557 Int32 thread_index;
558};
559}
560
561/*---------------------------------------------------------------------------*/
562/*---------------------------------------------------------------------------*/
563/*
564 * Cette fonction est celle appelée lors de la création d'un thread.
565 */
566void
567_ThreadWrapper(LaunchThreadInfo* lti)
568{
569 ArcaneMainBatch* amb = lti->arcane_main;
570 IApplication* main_app = lti->application;
571 ArcaneMainBatch::ExecFunctor functor(lti->session_exec,lti->thread_index);
572 bool clean_abort = false;
573 bool is_master = lti->thread_index == 0;
574 int r = ArcaneMain::callFunctorWithCatchedException(&functor,amb,&clean_abort,is_master);
575 if (r!=0 && !clean_abort){
576 // Le thread est terminé mais comme il est le seul à avoir planté,
577 // il est possible que les autres soient bloqués.
578 // Dans ce cas, on fait un abort pour éviter un blocage
579 // TODO: essayer de tuer les autres threads correctement.
580 if (main_app){
581 IParallelSuperMng* psm = main_app->parallelSuperMng();
582 psm->tryAbort();
583 }
584 }
585}
586
587/*---------------------------------------------------------------------------*/
588/*---------------------------------------------------------------------------*/
589
591execute()
592{
593 ITraceMng* trace = _application()->traceMng();
594
595 if (m_code_service.get())
596 m_session = m_code_service->createSession();
597
598 IParallelSuperMng* psm = _application()->parallelSuperMng();
599 Int32 nb_total_rank = psm->commSize();
600 const Integer nb_wanted_sub_domain = applicationBuildInfo().nbReplicationSubDomain();
601 CaseDatasetSource& dataset_source = _applicationBuildInfo().caseDatasetSource();
602 if (nb_wanted_sub_domain>nb_total_rank)
603 ARCANE_THROW(ArgumentException,"Number of subdomain '{0}' > number of allocated cores '{1}",
604 nb_wanted_sub_domain,nb_total_rank);
605
606 Integer nb_local_rank = psm->nbLocalSubDomain();
607 trace->info() << "NB_LOCAL_RANK=" << nb_local_rank;
608 if (nb_local_rank>=1)
610 int return_value = 0;
611
612 // Lecture des données du jeu de données.
613 if (dataset_source.content().empty() && m_direct_test_name.null()){
614 String case_file = dataset_source.fileName();
615 trace->info() << "Reading input data '" << case_file << "'";
616 IIOMng* io_mng = _application()->ioMng();
617 UniqueArray<Byte> case_bytes;
618 bool is_bad = io_mng->collectiveRead(case_file,case_bytes);
619 if (is_bad)
620 ARCANE_THROW(ParallelFatalErrorException,"Cannot read input data file '{0}'",case_file);
621 dataset_source.setContent(case_bytes);
622 }
623
624 m_session_exec = new SessionExec(this,m_session,nb_local_rank);
625
626 UniqueArray<LaunchThreadInfo> thinfo(nb_local_rank);
627 for( Integer i=0; i<nb_local_rank; ++i ){
628 thinfo[i].arcane_main = this;
629 thinfo[i].session_exec = m_session_exec;
630 thinfo[i].application = _application();
631 thinfo[i].thread_index = i;
632 }
633
634 if (nb_local_rank>1){
635 UniqueArray<std::thread*> gths(nb_local_rank);
636 for( Integer i=0; i<nb_local_rank; ++i ){
637 gths[i] = new std::thread(_ThreadWrapper,&thinfo[i]);
638 }
639 for( Integer i=0; i<nb_local_rank; ++i ){
640 gths[i]->join();
641 delete gths[i];
642 }
643 }
644 else{
646 m_session_exec->executeRank(0);
647 }
648
649 // TODO: supprimer car inutile car vaut toujours 0.
650 return return_value;
651}
652
653/*---------------------------------------------------------------------------*/
654/*---------------------------------------------------------------------------*/
655/*
656 * En mode avec un sous-domaine par thread,cette fonction est appelée
657 * par chaque thread (de manière potentiellement concurrente) pour son sous-domaine.
658 * \a local_rank indique le rang local du thread, qui est compris
659 * entre 0 et \a nb_local_sub_domain (tèl que définit dans execute()).
660 */
661void ArcaneMainBatch::SessionExec::
662executeRank(Int32 local_rank)
663{
664 // ATTENTION:
665 // Cette fonction doit etre reentrente...
666
667 auto sub_info = new SubInfo();
668 m_sub_infos[local_rank] = sub_info;
669
671 if (pas && m_has_sub_domain_threads){
672 // Il ne faut binder les CPU que si demandé et uniquement si
673 // le nombre de threads au total (sur l'ensemble des processus)
674 // ne dépasse pas le nombre de coeur de la machine.
675 if (!platform::getEnvironmentVariable("ARCANE_BIND_THREADS").null()){
676 ITraceMng* tm = _application()->traceMng();
677 tm->info() << "Binding threads";
678 pas->bindThread(local_rank);
679 }
680 }
681
682 // Création du gestionnaire de parallélisme pour l'ensemble des rangs alloués.
683 IParallelSuperMng* psm = _application()->parallelSuperMng();
684 Ref<IParallelMng> world_pm = psm->internalCreateWorldParallelMng(local_rank);
685 sub_info->m_world_parallel_mng = world_pm;
686
687 if (!m_direct_test_name.null()){
688 _execDirectTest(world_pm.get(),m_direct_test_name,true);
689 return;
690 }
691
692 // Regarde si on souhaite exécuter le calcul sur un sous-ensemble
693 // des ressources allouées. Pour l'instant, il est uniquement possible
694 // de choisir un nombre de sous-domaine. Si c'est le cas, seuls
695 // les rangs de 0 au nombre de sous-domaine souhaité moins 1 sont
696 // utilisés. Les rangs supérieurs n'ont pas de sous-domaines
697 // et à la place utilisent un service qui implémente IDirectExecution
698
699 // Création du gestionnaire des statistiques d'exécution.
700 ITraceMng* trace = world_pm->traceMng();
701 String stat_name = "Rank";
702 stat_name = stat_name + world_pm->commRank();
703 ITimeStats* time_stat = _application()->mainFactory()->createTimeStats(world_pm->timerMng(),trace,stat_name);
704 sub_info->m_time_stats = time_stat;
705 time_stat->beginGatherStats();
706 world_pm->setTimeStats(time_stat);
707
708 Ref<IParallelMng> pm = world_pm;
709 Ref<IParallelMng> all_replica_pm = pm;
710
711 const Integer nb_wanted_sub_domain = _application()->applicationBuildInfo().nbProcessusSubDomain();
712 const Integer nb_wanted_replication = _application()->applicationBuildInfo().nbReplicationSubDomain();
713 // On est en parallèle et on souhaite moins de sous-domaines que de processus alloués
714 if (world_pm->isParallel()){
715 // Pour l'instant, on ne peut pas mélanger la réplication de sous-domaines avec
716 // un nombre de sous-domaines différent du nombre de processeurs alloués.
717 // TODO: lorsque ce ne sera plus le cas, il faudra faire un all_replica_pm qui
718 // contiendra l'ensemble des sous-domaines et des réplica.
719
720 if (nb_wanted_replication>1){
721 Int32 comm_size = world_pm->commSize();
722 Int32 nb_sub_part = comm_size / nb_wanted_replication;
723 trace->info() << "Using sub-domain replication nb_sub_part=" << nb_sub_part;
724 if ((comm_size % nb_wanted_replication)!=0)
725 ARCANE_FATAL("The number of replication '{0}' must be a common factor of the number of allocated cores '{1}",
726 nb_wanted_replication,comm_size);
727 // D'abord, on créé un communicateur contenant les réplicats de chaque sous-domaine
728 // Ce communicateur contiendra donc \a m_nb_wanted_replication objets
729 Ref<IParallelMng> replicate_pm;
730 trace->info() << "Building replicated parallel mng";
731 {
732 Int32UniqueArray kept_ranks(nb_wanted_replication);
733 for( Integer i_sd=0; i_sd<nb_sub_part; ++i_sd ){
734 for( Int32 i=0; i<nb_wanted_replication; ++i ){
735 kept_ranks[i] = i_sd + (i*nb_sub_part);
736 trace->info() << "Rank r=" << kept_ranks[i];
737 }
738 Ref<IParallelMng> new_pm = world_pm->createSubParallelMngRef(kept_ranks);
739 if (new_pm.get()){
740 replicate_pm = new_pm;
741 replicate_pm->setTimeStats(time_stat);
742 trace->info() << " Building own replicated parallel mng";
743 }
744 else{
745 trace->info()<<"!pm";
746 }
747 trace->flush();
748 }
749 }
750 if (!replicate_pm)
751 ARCANE_FATAL("Null replicated parallel mng");
752
753 // Maintenant, on créé un IParallelMng qui correspond à l'ensemble
754 // des rangs d'un même réplica. Ce IParallelMng sera assigné au
755 // sous-domaine qui sera créé par la suite.
756 trace->info() << "Building sub-domain parallel mng";
757 {
758 Int32UniqueArray kept_ranks(nb_sub_part);
759 for( Integer i_repl=0; i_repl<nb_wanted_replication; ++i_repl ){
760 for( Int32 i=0; i<nb_sub_part; ++i ){
761 kept_ranks[i] = i + (i_repl*nb_sub_part);
762 trace->info() << "Rank r=" << kept_ranks[i];
763 }
764 Ref<IParallelMng> new_pm = world_pm->createSubParallelMngRef(kept_ranks);
765 if (new_pm.get()){
766 pm = new_pm;
767 if (nb_sub_part==1){
768 // Il faut prendre la version séquentielle pour faire comme si le calcul
769 // était séquentiel. Ce gestionnaire sera détruit en même temps
770 // que \a new_pm
771 pm = new_pm->sequentialParallelMngRef();
772 }
773 trace->info()<<"pm: setting time_stat & m_rank_parallel_mng for replica rank=" << i_repl;
774 trace->flush();
775 pm->setTimeStats(time_stat);
776 sub_info->m_rank_parallel_mng = new_pm;
777 auto pr = new ParallelReplication(i_repl,nb_wanted_replication,replicate_pm);
778 pm->setReplication(pr);
779 }
780 else{
781 trace->info()<<"!pm";
782 trace->flush();
783 }
784 }
785 }
786 }
787 else if (nb_wanted_sub_domain!=0){
788 const Int32 nb_sub_part = nb_wanted_sub_domain;
789 Int32UniqueArray kept_ranks(nb_sub_part);
790 for( Int32 i=0; i<nb_sub_part; ++i )
791 kept_ranks[i] = i;
792 pm = world_pm->createSubParallelMngRef(kept_ranks);
793 if (pm.get()){
794 trace->info()<<"pm: setting time_stat & m_rank_parallel_mng";
795 trace->flush();
796 pm->setTimeStats(time_stat);
797 sub_info->m_rank_parallel_mng = pm;
798 all_replica_pm = pm;
799 }
800 else{
801 trace->info()<<"!pm";
802 trace->flush();
803 }
804 }
805 }
806
807 bool print_stats = false;
808 ISubDomain* sub_domain = nullptr;
809
810 if (!pm){
811 // Si ici, il s'agit d'un rang qui ne possède pas de sous-domaine.
812 // Dans ce cas, exécute le service donnée par 'm_idle_service_name'
813 // (si spécifié, sinon ne fait rien)
814 trace->info()<<"The rank doesn't own any subdomain!";
815 if (m_properties.m_idle_service_name.empty()){
816 trace->info() << "No idle service specified"; trace->flush();
817 }
818 else{
819 trace->info()<<"execDirectTest: "<< m_properties.m_idle_service_name;
820 trace->flush();
821 _execDirectTest(world_pm.get(),m_properties.m_idle_service_name,false);
822 // On sort de l'execute() du directTest grâce au broadcast(This is the end), il faut s'en retourner
823 return;
824 }
825 print_stats = true;
826 }
827 else {
828 _createAndRunSubDomain(sub_info,pm,all_replica_pm,local_rank);
829 sub_domain = sub_info->m_sub_domain;
830 print_stats = sub_info->m_want_print_stats;
831 }
832
833 time_stat->endGatherStats();
834
835 if (print_stats && sub_domain){
836 // S'assure que tout le monde est ici avant d'arêter le profiling
837 // TODO: Comme le profiling est local au processus, il suffirait
838 // a priori de faire la barrière sur les IParallelMng locaux.
839 IParallelMng* pm = sub_domain->parallelMng();
840 pm->barrier();
841 if (local_rank==0)
843 pm->barrier();
844 _printStats(sub_domain,trace,time_stat);
845
846 // On doit détruire les variables en mémoire partagée ici parce que leur
847 // destruction est effectuée collectivement.
848 // On ne peut pas détruire toutes les variables car certaines sont
849 // utilisées après (GlobalIteration par exemple).
850 // Si, un jour, on met certaines variables "Global" en mémoire partagée,
851 // cette partie va poser problème.
852 sub_domain->variableMng()->_internalApi()->removeAllShMemVariables();
853 }
854
855 //BaseForm[Hash["This is the end", "CRC32"], 16]
856 // On informes les 'autres' capacités qu'il faut s'en aller, maintenant!
857 world_pm->broadcast(UniqueArray<unsigned long>(1,0xdfeb699fl).view(),0);
858}
859
860/*---------------------------------------------------------------------------*/
861/*---------------------------------------------------------------------------*/
862
863void ArcaneMainBatch::SessionExec::
864_createAndRunSubDomain(SubInfo* sub_info,Ref<IParallelMng> pm,Ref<IParallelMng> all_replica_pm,Int32 local_rank)
865{
866 // Il s'agit d'un rang qui a un sous-domaine.
867 // Celui-ci est créé et l'exécution commence.
868 SubDomainBuildInfo sdbi(pm,local_rank,all_replica_pm);
869 sdbi.setCaseFileName(m_case_file);
870 sdbi.setCaseContent(m_case_bytes);
871 ISubDomain* sub_domain = m_code_service->createAndLoadCase(m_session,sdbi);
872 sub_info->m_sub_domain = sub_domain;
873
874 ITraceMng* trace = _application()->traceMng();
875 ITraceMng* sd_trace = sub_domain->traceMng();
876 ITraceMngPolicy* trace_policy = _application()->getTraceMngPolicy();
877
878 // En cas de réplication, désactive les sorties courbes
879 // des réplicats.
880 trace->info() << "REPLICATION: rank=" << pm->replication()->replicationRank();
881
882 if (!pm->replication()->isMasterRank()){
883 trace->info() << "Disable output curves for replicates.";
884 sub_domain->timeHistoryMng()->setDumpActive(false);
885 }
886
887 // TODO:
888 // Détruire le sous-domaine à la fin de la fonction mais il
889 // faut pour cela modifier ISession pour supporter la suppression
890 // d'un sous-domaine (et ensuite détruire ISession).
891
892 IProcessorAffinityService* pas = platform::getProcessorAffinityService();
893 if (pas){
894 String cpu_set = pas->cpuSetString();
895 trace->info() << " CpuSet=" << cpu_set;
896 }
897
898 if (m_arcane_main->m_check_case_only){
899 trace->info() << "Checking the input data";
900 // Initialise les modules de la boucle en temps
901 {
902 TimeLoopReader stl(_application());
903 stl.readTimeLoops();
904 stl.registerTimeLoops(sub_domain);
905 stl.setUsedTimeLoop(sub_domain);
906 }
907 ICaseMng* cm = sub_domain->caseMng();
908 cm->readOptions(true);
909 }
910 else{
911 Timer init_timer(sub_domain,"InitTimer",Timer::TimerReal);
912 Timer loop_timer(sub_domain,"LoopTimer",Timer::TimerReal);
913
914 {
915 Timer::Action ts_action(sub_domain,"Init");
916 Timer::Sentry ts(&init_timer);
917
918 m_code_service->initCase(sub_domain,m_properties.m_is_continue);
919 }
920
921 if (m_properties.m_max_iteration>0)
922 trace->info() << "Option 'max_iteration' activated with " << m_properties.m_max_iteration;
923
924 // Redirige les signaux.
925 // Cela se fait aussi a l'initialisation mais ici on peut être dans un autre
926 // thread et de plus certaines bibliothèques ont pu rediriger les signaux
927 // lors de l'init
928 {
929 CriticalSection cs(pm->threadMng());
930 ArcaneMain::redirectSignals();
931 }
932 int ret_compute_loop = 0;
933
934 IDirectExecution* direct_exec = sub_domain->directExecution();
935 if (direct_exec && direct_exec->isActive()){
936 trace->info() << "Direct execution activated";
937 direct_exec->execute();
938 }
939 else if (m_arcane_main->m_init_only){
940 trace->info() << "Option 'init_only' activated";
941 sub_info->m_want_print_stats = true;
942 }
943 else{
944 sub_info->m_want_print_stats = true;
945 Timer::Action ts_action(sub_domain,"Loop");
946 Timer::Sentry ts(&loop_timer);
947 // Lors de la boucle de calcul, ne force pas l'affichage des traces à un niveau
948 // donné (ce qui est fait lors de l'initialisation de l'application.
949 trace_policy->setDefaultVerboseLevel(sd_trace,Trace::UNSPECIFIED_VERBOSITY_LEVEL);
950 if (m_direct_sub_domain_execute_functor){
951 m_direct_sub_domain_execute_functor->setSubDomain(sub_domain);
952 m_direct_sub_domain_execute_functor->execute();
953 sub_domain->parallelMng()->barrier();
954 }
955 else{
956 ret_compute_loop = sub_domain->timeLoopMng()->doComputeLoop(m_properties.m_max_iteration);
957 if (ret_compute_loop<0)
958 //TODO: NE PAS REMPLIR DIRECTEMENT CETTE FONCTION CAR CELA NE MARCHE
959 // PAS EN MULTI-THREAD
960 m_arcane_main->setErrorCode(8);
961 }
962 }
963 {
964 Real init_time = init_timer.totalTime();
965 Real loop_time = loop_timer.totalTime();
966 trace->info(0) << "TotalReel = " << (init_time+loop_time)
967 << " secondes (init: "
968 << init_time << " loop: " << loop_time << " )";
969 }
970 {
971 Timer::Action ts_action(sub_domain,"Exit");
972 trace_policy->setDefaultVerboseLevel(sd_trace,Trace::DEFAULT_VERBOSITY_LEVEL);
973 sub_domain->doExitModules();
974 }
975 }
976}
977
978/*---------------------------------------------------------------------------*/
979/*---------------------------------------------------------------------------*/
980
981void ArcaneMainBatch::SessionExec::
982_printStats(ISubDomain* sub_domain,ITraceMng* trace,ITimeStats* time_stat)
983{
984 ExecutionStatsDumper exec_dumper(trace);
985 exec_dumper.dumpStats(sub_domain,time_stat);
986}
987
988/*---------------------------------------------------------------------------*/
989/*---------------------------------------------------------------------------*/
990
991void ArcaneMainBatch::SessionExec::
992_execDirectTest(IParallelMng* world_pm,const String& test_name,bool is_collective)
993{
994 ITraceMng* trace = world_pm->traceMng();
995 trace->info() << "Direct test name=" << test_name;
996 trace->flush();
997 ServiceFinder2T<IDirectExecution,IApplication> sf(_application(),_application());
998 Ref<IDirectExecution> exec(sf.createReference(test_name));
999 if (!exec){
1000 String msg = String::format("Can not find 'IDirectExecution' service name '{0}'",test_name);
1001 if (is_collective)
1002 throw ParallelFatalErrorException(A_FUNCINFO,msg);
1003 else
1004 throw FatalErrorException(A_FUNCINFO,msg);
1005 }
1006 else{
1007 trace->info() << "Begin execution of direct service";
1008 trace->flush();
1009 }
1010 exec->setParallelMng(world_pm);
1011 exec->execute();
1012}
1013
1014/*---------------------------------------------------------------------------*/
1015/*---------------------------------------------------------------------------*/
1016
1018finalize()
1019{
1020 if (m_session){
1021 m_session->endSession(errorCode());
1022 _application()->removeSession(m_session);
1023 delete m_session;
1024 m_session = nullptr;
1025 }
1026 m_code_service.reset();
1027 delete m_session_exec;
1028 m_session_exec = nullptr;
1029
1030 ITraceMng* tm = _application()->traceMng();
1032}
1033
1034/*---------------------------------------------------------------------------*/
1035/*---------------------------------------------------------------------------*/
1036
1038doAbort()
1039{
1040 if (m_session)
1041 m_session->doAbort();
1042 else{
1043 // Pour finir proprement même si arrêt avant la création de la session
1044 // ou après la destruction de la session.
1046 if (psm)
1047 psm->tryAbort();
1048 }
1049}
1050
1051/*---------------------------------------------------------------------------*/
1052/*---------------------------------------------------------------------------*/
1053
1054template<typename V> void ArcaneMainBatchProperties::
1055_applyPropertyVisitor(V& p)
1056{
1057 auto b = p.builder();
1058
1059 p << b.addInt32("MaxIteration")
1060 .addDescription("Maximum number of iteration")
1061 .addCommandLineArgument("MaxIteration")
1062 .addGetter([](auto a) { return a.x.m_max_iteration; })
1063 .addSetter([](auto a) { a.x.m_max_iteration = a.v; });
1064
1065 p << b.addBool("Continue")
1066 .addDescription("True if continue from previous execution (restart)")
1067 .addCommandLineArgument("Continue")
1068 .addGetter([](auto a) { return a.x.m_is_continue; })
1069 .addSetter([](auto a) { a.x.m_is_continue = a.v; });
1070
1071 p << b.addString("IdleService")
1072 .addDescription("Name of the idle service for additionnal cores")
1073 .addCommandLineArgument("IdleService")
1074 .addGetter([](auto a) { return a.x.m_idle_service_name; })
1075 .addSetter([](auto a) { a.x.m_idle_service_name = a.v; });
1076}
1077
1078/*---------------------------------------------------------------------------*/
1079/*---------------------------------------------------------------------------*/
1080
1081ARCANE_REGISTER_PROPERTY_CLASS(ArcaneMainBatchProperties,());
1082
1083/*---------------------------------------------------------------------------*/
1084/*---------------------------------------------------------------------------*/
1085
1086} // End namespace Arcane
1087
1088/*---------------------------------------------------------------------------*/
1089/*---------------------------------------------------------------------------*/
#define ARCANE_THROW(exception_class,...)
Macro pour envoyer une exception avec formattage.
#define ARCANE_FATAL(...)
Macro envoyant une exception FatalErrorException.
static void finalize(ITraceMng *tm)
Finalise l'exécution.
Definition Runner.cc:517
static void stopAllProfiling()
Stoppe toutes les activités de profiling.
Definition Runner.cc:505
CaseDatasetSource & caseDatasetSource()
Source du jeu de données.
Informations sur une application.
Propriétés associées à ArcaneMain.
String m_idle_service_name
Nom du service pour les CPU non utilisés.
void executeFunctor() override
Exécute la méthode associé
Infos par sous-domaine qui doivent être détruites à la fin de l'exécution.
Informations d'exécution pour une session.
Ref< ICodeService > m_code_service
Service du code.
bool m_has_sub_domain_threads
indique si on utilise des threads pour gérer des sous-domaines
UniqueArray< std::byte > m_case_bytes
Contenu du jeu de données du cas sous forme d'un document XML.
String m_case_file
Nom du fichier contenant le cas.
const ArcaneMainBatchProperties m_properties
Propriétés d'exécution.
Exécution en mode batch d'un code.
String m_case_name
Nom du cas.
void doAbort() override
Effectue un abort.
Ref< ICodeService > m_code_service
Service du code.
bool m_init_only
true si on ne fait que l'initialisation.
ArcaneMainBatchProperties m_properties
Session.
void initialize() override
Initialise l'instance. L'instance n'est pas utilisable tant que cette méthode n'a pas été appelée.
int execute() override
Lance l'exécution. Cette méthode ne retourne que lorsqu'on quitte le programme.
bool m_check_case_only
true si on ne fait que vérifier le jeu de données.
void finalize() override
Effectue les dernières opérations avant destruction de l'instance.
bool m_has_sub_domain_threads
indique si on utilise des threads pour gérer des sous-domaines
void build() override
Construit les membres la classe. L'instance n'est pas utilisable tant que cette méthode n'a pas été a...
bool parseArgs(StringList args) override
Analyse les arguments.
const ApplicationInfo & applicationInfo() const override
Informations sur l'éxécutable.
static int callFunctorWithCatchedException(IFunctor *functor, IArcaneMain *amain, bool *clean_abort, bool is_print=true)
Appelle le fonctor functor en récupérant les éventuelles exceptions.
IApplication * application() const override
Application.
Definition ArcaneMain.h:321
void build() override
Construit les membres la classe. L'instance n'est pas utilisable tant que cette méthode n'a pas été a...
const ApplicationBuildInfo & applicationBuildInfo() const override
Informations pour construire l'instance IApplication.
int errorCode() const override
Code d'erreur de l'exécution.
Definition ArcaneMain.h:311
void initialize() override
Initialise l'instance. L'instance n'est pas utilisable tant que cette méthode n'a pas été appelée.
bool parseArgs(StringList args) override
Analyse les arguments.
Exception lorsqu'un argument est invalide.
Source d'un jeu de données d'un cas.
void setFileName(const String &name)
Positionne le nom du fichier du jeu de données.
String fileName() const
Nom du fichier du jeu de données.
void setContent(Span< const std::byte > bytes)
Positionne le contenu du jeu de données.
ByteConstSpan content() const
Contenu du jeu de données.
Integer count() const
Nombre d'éléments de la collection.
Interface de l'application.
virtual IParallelSuperMng * parallelSuperMng()=0
Gestionnaire superviseur du parallélisme.
Interface de la classe de gestion du code.
Definition IArcaneMain.h:54
virtual ITraceMng * traceMng() const =0
Gestionnaire de traces.
Interface d'un fonctor pour exécuter du code directement après la création d'un sous-domaine sans pas...
Interface du gestionnaire des entrées sorties.
Definition IIOMng.h:36
virtual bool collectiveRead(const String &filename, ByteArray &bytes)=0
Lecture collective d'un fichier.
Manufacture des classes d'Arcane.
Interface du gestionnaire de parallélisme pour un sous-domaine.
Classe abstraite du superviseur de parallélisme.
virtual void tryAbort()=0
Tente de faire un abort.
virtual Int32 nbLocalSubDomain()=0
Nombre de sous-domaines à créér localement.
virtual Int32 commSize() const =0
Retourne le nombre total de process utilisés.
Interface d'un service de gestion de l'affinité des coeurs CPU.
virtual void bindThread(Int32 cpu)=0
Contraint le thread courant à rester sur le coeur d'indice cpu.
Interface d'une session d'exécution d'un cas.
Definition ISession.h:37
Interface du gestionnaire d'un sous-domaine.
Definition ISubDomain.h:74
Interface gérant les statistiques sur les temps d'exécution.
Definition ITimeStats.h:43
Interface du gestionnaire de traces.
virtual TraceMessage pfatal()=0
Flot pour un message d'erreur fatale parallèle.
virtual TraceMessage info()=0
Flot pour un message d'information.
Exception lorsqu'une erreur fatale 'parallèle' est générée.
Référence à une instance.
Chaîne de caractères unicode.
bool empty() const
Vrai si la chaîne est vide (nulle ou "")
Definition String.cc:316
@ TimerReal
Timer utilisant le temps réel.
Definition Timer.h:76
Vecteur 1D de données avec sémantique par valeur (style STL).
Integer toInteger(const char *str, bool *is_ok=0)
Converti la chaîne str en un entier non signé. Si is_ok n'est pas nul, il vaut true en retour si la c...
ARCCORE_BASE_EXPORT String getEnvironmentVariable(const String &name)
Variable d'environnement du nom name.
IProcessorAffinityService * getProcessorAffinityService()
Service utilisé pour la gestion de l'affinité des processeurs.
-*- tab-width: 2; indent-tabs-mode: nil; coding: utf-8-with-signature -*-
Int32 Integer
Type représentant un entier.
List< String > StringList
Tableau de chaînes de caractères unicode.
Definition UtilsTypes.h:509
UniqueArray< Int32 > Int32UniqueArray
Tableau dynamique à une dimension d'entiers 32 bits.
Definition UtilsTypes.h:341
double Real
Type représentant un réel.
std::int32_t Int32
Type entier signé sur 32 bits.