Класс, опеделяющий список решаемых задач и очередь их прохождения More...

#include <Queue.h>

Collaboration diagram for VMlib::Queue:

[legend]

Public Member Functions
	Queue (int &argc, char *&argv, void(_CreateMpiTypes)())
	Конструктор More...

	~Queue ()
	Деструктор More...

void	TaskSplit ()
	Процедура постановка новых задач на отсчет и занятие процессоров More...

void	TaskUpdate ()
	Процедура обновления состояния задач и процессоров More...

void	RunConveyer ()
	Запуск вычислительного конвейера (в рамках кванта времени) More...

void	LoadTasksList (const std::string &_tasksFile, const std::string &_mechanicsFile, const std::string &_defaultsFile, const std::string &_switchersFile)
	Загрузка списка задач More...

Public Attributes
std::vector< Task >	task
	Список описаний решаемых задач More...

std::unique_ptr< WorldGen >	world
	Умный указатель на текущую решаемую задачу More...

int	myProcState
	Состояние данного процессора More...

int	myProcStateVar
	Состояние данного процессора More...

const double	kvantTime = 10.0
	Продолжительность кванта времени в секундах More...

int	currentKvant
	Номер текущего кванта времени More...

int	nextKvant
	Признак необходимости выполнения следующего кванта и продолжения расчета More...

Parallel	parallel
	Класс, опеделяющий параметры исполнения задачи в параллельном MPI-режиме More...

int	prepared
	Число подготовленных к запуску задач More...

int	solving
	Число решаемых в данный момент задач More...

int	finished
	Число уже решенных задач More...

Private Member Functions
void	ConstructProcStateVar ()
	Процедура, нумерующая задачи в возрастающем порядке More...

void	AddTask (int _nProc, std::unique_ptr< PassportGen > _passport)
	Добавление задачи в список More...

Private Attributes
struct {
int prepared
	Число подготовленных к запуску задач More...

int solving
	Число решаемых в данный момент задач More...

int finished
	Число уже решенных задач More...

}	numberOfTask
	Структура, содержащая информацию о количестве задач в данный момент времени More...

std::vector< int >	procState
	Список состояний процессоров More...

std::vector< int >	procStateVar
	Модифицированный список состояний процессоров More...

std::vector< int >	flagFinish
	Список возвращаемых флагов останова задачи More...

int	myidAll
	Глобальный номер процессора More...

int	nProcAll
	Общее число процессоров More...

int	groupAll

int	groupStarting

int	commStarting

int	groupSolving

int	commSolving

int	sizeCommSolving
	Число процессоров в группе для головных процессоров в решаемых в данном кванте времени задачах More...

LogStream	info
	Поток для вывода логов и сообщений об ошибках More...

Detailed Description

Класс, опеделяющий список решаемых задач и очередь их прохождения

Управляет распределением задач по процессорам, инициализацией их запуска и выгрузки

Author: Марчевский Илья Константинович 1.12

Date: 14 января 2024 г.

Definition at line 72 of file Queue.h.

Constructor & Destructor Documentation

Queue::Queue	(	int &	argc,
		char **&	argv,
		void(*)()	_CreateMpiTypes
	)

Конструктор

Производит инициализацию очереди перед запуском всего процесса

Parameters

[in]	argc	ссылка на число параметров командной строки
[in]	argv	ссылка на указатель на список параметров командной строки
[in]	_CreateMpiTypes	указатель на функцию, инициализирующую все необходимые MPI-описания типов

Definition at line 75 of file Queue.cpp.

 {
 #ifdef USE_MPI
     MPI_Init(&argc, &argv);
     MPI_Comm_size(MPI_COMM_WORLD, &nProcAll);
     MPI_Comm_rank(MPI_COMM_WORLD, &myidAll);
     MPI_Comm_group(MPI_COMM_WORLD, &groupAll);
     _CreateMpiTypes();  
 #else
     nProcAll = 1;
     myidAll = 0;
     groupAll = -1;
 #endif
 
     //готовим систему к запуску
     //(выполняется на глобальном нулевом процессоре)
     if (myidAll == 0)
     {
         PrintUniversalLogoToStream(*defaults::defaultWorld2DLogStream);
 
         info.assignStream(defaults::defaultQueueLogStream, "queue");
 
         //Устанавливаем флаг занятости в состояние "свободен" всем процессорам,     
 #ifdef USE_MPI
         procState.resize(nProcAll, MPI_UNDEFINED);
         procStateVar.resize(nProcAll, MPI_UNDEFINED);
 #else
         procState.resize(nProcAll, -32766);
         procStateVar.resize(nProcAll, -32766);
 #endif
 
         numberOfTask.solving  = 0; //число решаемых в данный момент задач
         numberOfTask.prepared = 0; //число подготовленных к запуску задач (в том числе)
         numberOfTask.finished = 0; //число уже отсчитанных задач    
 
         //текущий номер кванта времени
         currentKvant = -1;
     }//if myid==0 
 }//Queue()

Here is the call graph for this function:

Queue::~Queue ( )

Деструктор

Definition at line 117 of file Queue.cpp.

 {
     info('i') << "Goodbye!" << std::endl;
 #ifdef USE_MPI
     MPI_Finalize();
 #endif
 }//~Queue()

Member Function Documentation

void Queue::AddTask	(	int	_nProc,
		std::unique_ptr< PassportGen >	_passport
	)

private

Добавление задачи в список

Parameters

[in]	_nProc	число запрашиваемых процессоров
[in]	_passport	умный указатель на паспорт задачи, направляемой в очередь

Definition at line 708 of file Queue.cpp.

 {
     task.resize(task.size() + 1);
     (task.end() - 1)->nProc = _nProc;
     (task.end() - 1)->passport = std::move(_passport);
     (task.end() - 1)->proc.resize(_nProc);
     (task.end() - 1)->state = TaskState::waiting;
 }//AddTask(...)

Here is the caller graph for this function:

void Queue::ConstructProcStateVar ( )

private

Процедура, нумерующая задачи в возрастающем порядке

Необходима для корректного распределения задач по процессорам

Definition at line 622 of file Queue.cpp.

 {
     std::vector<bool> prFlag;
     int number = -1;
 
     prFlag.resize(nProcAll, true); //их надо просматривать
 
     for (int i = 0; i<nProcAll; ++i)
 
 #ifdef USE_MPI
     if (procState[i] == MPI_UNDEFINED)
     {
         prFlag[i] = false;  //уже просмотрели
         procStateVar[i] = MPI_UNDEFINED;
     } //if (ProcState[i]==MPI_UNDEFINED)
 #else
     if (procState[i] == -32766)
     {
         prFlag[i] = false;  //уже просмотрели
         procStateVar[i] = -32766;
     } //if (ProcState[i]==-32766)
 #endif
 
     for (int i = 0; i<nProcAll; ++i)
     if (prFlag[i])
     {
         prFlag[i] = false;
         number++;
         
         for (int s = i; s<nProcAll; ++s)
         if (procState[s] == procState[i])
         {
             procStateVar[s] = number;
             prFlag[s] = false;
         }//if (ProcState[s]==ProcState[i])
     } //if (pr_flag[i])                                 
 }//ConstructProcStateVar()

Here is the caller graph for this function:

void Queue::LoadTasksList	(	const std::string &	_tasksFile,
		const std::string &	_mechanicsFile,
		const std::string &	_defaultsFile,
		const std::string &	_switchersFile
	)

Загрузка списка задач

Parameters

[in]	_tasksFile	константная ссылка на имя файла с описанием очереди задач
[in]	_mechanicsFile	константная ссылка на имя файла со словарем механических систем
[in]	_defaultsFile	константная ссылка на имя файла с описанием параметров по умолчанию
[in]	_switchersFile	константная ссылка на имя файла со значениями параметров-переключателей

Definition at line 719 of file Queue.cpp.

 {
     std::string extTasksFile = _tasksFile;
     std::string extMechanicsFile = _mechanicsFile;
     std::string extDefaultsFile = _defaultsFile;
     std::string extSwitchersFile = _switchersFile;
 
     if (
         fileExistTest(extTasksFile, info, {"txt", "TXT"}) &&
         fileExistTest(extDefaultsFile, info, { "txt", "TXT" }) &&
         fileExistTest(extSwitchersFile, info, { "txt", "TXT" })
     )
     {
         std::stringstream tasksFile(Preprocessor(extTasksFile).resultString);
         std::stringstream defaultsFile(Preprocessor(extDefaultsFile).resultString);
         std::stringstream switchersFile(Preprocessor(extSwitchersFile).resultString);
         std::vector<std::string> taskFolder;
 
         std::unique_ptr<StreamParser> parserTaskList;
         parserTaskList.reset(new StreamParser(info, "parser", tasksFile, '(', ')'));
 
         std::vector<std::string> alltasks;
         parserTaskList->get("problems", alltasks);
 
         std::ptrdiff_t nTasks = std::count_if(alltasks.begin(), alltasks.end(), [](const std::string& a) {return (a.length() > 0);});
         info.endl();
         info('i') << "Number of problems to be solved: " << nTasks << std::endl;
         
         for (size_t i = 0; i < alltasks.size(); ++i)
         {
             if (alltasks[i].length() > 0)
             {
                 //делим имя задачи + выражение в скобках на 2 подстроки
                 std::pair<std::string, std::string> taskLine = StreamParser::SplitString(info, alltasks[i], false);
 
                 std::string dir = taskLine.first;
 
                 info.endl();
                 info('i') << "-------- Loading problem #" << i << " (" << dir << ") --------" << std::endl;
 
                 //вторую подстроку разделяем на вектор из строк по запятым, стоящим вне фигурных скобок     
                 std::vector<std::string> vecTaskLineSecond = StreamParser::StringToVector(taskLine.second, '{', '}');
 
                 //создаем парсер и связываем его с параметрами профиля
                 std::stringstream tskStream(StreamParser::VectorStringToString(vecTaskLineSecond));
                 std::unique_ptr<StreamParser> parserTask;
 
                 parserTask.reset(new StreamParser(info, "problem parameters", tskStream, defaultsFile, {"pspfile", "np", "copyPath"}));
 
                 std::string pspFile;
                 int np;             
 
                 //считываем нужные параметры с учетом default-значений
                 parserTask->get("pspfile", pspFile, &defaults::defaultPspFile);
                 parserTask->get("np", np, &defaults::defaultNp);                
 
                 if (np > 1)  // 31/12/2023
                 {
                     info.endl();
                     info('e') << "Internal MPI-parallelization is not supported longer!" << std::endl;
                 }
 
                 std::string copyPath;
                 parserTask->get("copyPath", copyPath, &defaults::defaultCopyPath);
                 if (copyPath.length() > 0)
                 {
                     //Копировать паспорт поручаем только одному процессу
                     if (myidAll == 0)
                     {
                         std::string initDir(copyPath.begin(), copyPath.end());
                         std::string command;
 
 #if defined(_WIN32)
                         std::replace(dir.begin(), dir.end(), '/', '\\');
                         std::replace(initDir.begin(), initDir.end(), '/', '\\');
 #endif
 
                         VMlib::CreateDirectory(dir.c_str(), "");
 
 #if defined(_WIN32)     
                         command = "copy \"" + initDir + "\\*.*\" "+ dir + "\\";
 #else
                         command = "cp ./" + initDir + "/* " + dir + "/";
 #endif
 
                         std::cout << "Copying files from folder \"" << initDir << "\" to \"" << dir << "\"" << std::endl;
                         
                         int systemRet = system(command.c_str());
                         if(systemRet == -1)
                         {
                             // The system method failed
                             info('e') << "problem #" << i << " (" << dir << \
                                          ") copying passport system method failed" << std::endl;
                             exit(-1);
                         }                   
                         std::cout << "Copying OK " << std::endl << std::endl;
 
                         //copyFile(pspFile, "./" + dir + "/" + pspFile);
                     }
                 }
 
 #ifdef USE_MPI              
                 MPI_Barrier(MPI_COMM_WORLD);
 #endif
 
                 std::unique_ptr<PassportGen> ptrPsp;
                 
 #ifdef CODE2D               
                 ptrPsp.reset(new VM2D::Passport(info, dir, i, pspFile, extMechanicsFile, extDefaultsFile, extSwitchersFile, vecTaskLineSecond));
 #endif
 
 #ifdef CODE3D           
                 ptrPsp.reset(new VM3D::Passport(info, dir, i, pspFile, extMechanicsFile, extDefaultsFile, extSwitchersFile, vecTaskLineSecond));
 #endif              
                                 
                 AddTask(np, std::move(ptrPsp));
 
                 info('i') << "-------- Problem #" << i << " (" << dir << ") is loaded --------" << std::endl << std::endl;
             } 
         } //for i
 
         //tasksFile.close();
         tasksFile.clear();
     
         //defaultsFile.close();
         defaultsFile.clear();
 
         //switchersFile.close();
         switchersFile.clear();
 
     }
     else
     {
         exit(1);
     }
 
 }

Here is the call graph for this function:












Here is the caller graph for this function:

void Queue::RunConveyer ( )

Запуск вычислительного конвейера (в рамках кванта времени)

Definition at line 662 of file Queue.cpp.

 {   
 #ifdef USE_MPI
     if (parallel.commWork != MPI_COMM_NULL)
 #else
     if (parallel.commWork != 0x04000000)
 #endif
     {
         double kvantStartWallTime; //Время на главном процессоре, когда начался очередной квант
         double deltaWallTime;
 
         if (parallel.myidWork == 0)
 #ifdef USE_MPI
             kvantStartWallTime = MPI_Wtime();
 #else
             kvantStartWallTime = omp_get_wtime();
 #endif
 
         //if (world->getCurrentStep() == 0)
         //  world->ZeroStep();
 
         do
         {
             if (!world->isFinished())
                 world->Step();
                     
             if (parallel.myidWork == 0)
 #ifdef USE_MPI
                 deltaWallTime = MPI_Wtime() - kvantStartWallTime;
 #else
                 deltaWallTime = omp_get_wtime() - kvantStartWallTime;
 #endif
             
 #ifdef USE_MPI
             MPI_Bcast(&deltaWallTime, 1, MPI_DOUBLE, 0, parallel.commWork);     
 #endif
         }
         //Проверка окончания кванта по времени (или завершения задачи)
         while (deltaWallTime < kvantTime);
 
     }//if (commWork != MPI_COMM_NULL)
 }//RunConveyer()

Here is the caller graph for this function:

void Queue::TaskSplit ( )

Процедура постановка новых задач на отсчет и занятие процессоров

Выполняется в начале очередного кванта

Definition at line 127 of file Queue.cpp.

 {       
     //Пересылаем информацию о состоянии процессоров на все компьютеры
     
 #ifdef USE_MPI
     MPI_Scatter(procState.data(), 1, MPI_INT, &myProcState, 1, MPI_INT, 0, MPI_COMM_WORLD);
 #else
     myProcState = procState[0];
 #endif
         
 #ifdef USE_MPI
     if (myProcState == MPI_UNDEFINED)
         world.reset(nullptr);
 #else
     if (myProcState == -32766)
         world.reset(nullptr);
 #endif
     
     if (myidAll == 0)
     {
         currentKvant++;
 
         int nfree = 0; //число свободных процессоров
 
         //Обнуляем число готовых к старту задач
         numberOfTask.prepared = 0;
 
         //считаем число свободных процессоров
         for (int i = 0; i < nProcAll; ++i)
 #ifdef USE_MPI
         if (procState[i] == MPI_UNDEFINED)
 #else
         if (procState[i] == -32766)
 #endif
             nfree++;
     
         //формируем номер следующей задачи, которая возможно будет поставлена на отсчет:
         size_t taskFol = numberOfTask.finished + numberOfTask.solving;
 
         //проверяем, достаточно ли свободных процессоров для запуска еще одной задачи
         //при условии, что не все задачи уже решены
         while ((taskFol < task.size()) && (nfree >= task[taskFol].nProc))
         {
             int p = 0; //число найденных свободных процессоров под задачу
             int j = 0; //текущий счетчик процессоров
 
             //подбираем свободные процессоры
             do
             {
 #ifdef USE_MPI
                 if (procState[j] == MPI_UNDEFINED)
 #else
                 if (procState[j] == -32766)
 #endif
                 {
                     //состояние процессора устанавливаем в "занят текущей задачей"
                     procState[j] = static_cast<int>(taskFol);
 
                     //номер процессора посылаем в перечень процессоров, решающих текущую задачу
                     task[taskFol].proc[p] = j;
 
                     //увеличиваем число найденных процессоров на единицу
                     p++;
                 }//if (ProcState[j]==MPI_UNDEFINED)
 
                 //переходим к следующему процессору
                 j++;
             } while (p < task[taskFol].nProc);          
 
             //состояние задачи устанавливаем в режим "стартует"
             task[taskFol].state = TaskState::starting;
 
             //отмечаем номер кванта, когда задача начала считаться
             task[taskFol].startEndKvant.first = currentKvant;
 
             //изменяем счетчики числа задач
             numberOfTask.prepared++;
             numberOfTask.solving++;
 
             //изменяем счетчик свободных процессоров
             nfree -= task[taskFol].nProc;
 
             //переходим к следующей задаче
             taskFol++;
 
         }//while ((nfree >= Task[task_fol].nproc)&&(task_fol<Task.size()))
 
     }//if (myidAll == 0)
     
     //Вывод информации о занятости процессоров задачами
     if (myidAll == 0)
     {
         info.endl();
         info('i') << "ProcStates: " << std::endl;
         for (int i = 0; i < nProcAll; ++i)
             info('-') << "proc[" << i << "] <=> problem[" << procState[i] << "]" << std::endl;
         info.endl();
     }
 
     //Пересылаем информацию о состоянии процессоров на все компьютеры
 #ifdef USE_MPI
     MPI_Scatter(procState.data(), 1, MPI_INT, &myProcState, 1, MPI_INT, 0, MPI_COMM_WORLD);
 #else
     myProcState = procState[0];
 #endif
 
     //Синхронизируем информацию
 #ifdef USE_MPI
     MPI_Bcast(&numberOfTask.solving, 1, MPI_INT, 0, MPI_COMM_WORLD);
     MPI_Bcast(&numberOfTask.prepared, 1, MPI_INT, 0, MPI_COMM_WORLD);
     MPI_Bcast(&numberOfTask.finished, 1, MPI_INT, 0, MPI_COMM_WORLD);
 #endif
     
     //список головных процессоров на тех задачах, 
     //которые только что поставлены на счет:
     std::vector<int> prepList;
     
     //формируем списки prepList и pspList
     //(выполняется на глобальном нулевом процессоре)
     if (myidAll == 0)
     {
         for (size_t i = 0; i<task.size(); ++i)
         //находим задачи в состоянии "стартует"
         if (task[i].state == TaskState::starting)
         {
             //запоминаем номер того процессора, что является там головным
             prepList.push_back(task[i].proc[0]);
             
             //состояние задачи вереводим в "считает"
             task[i].state = TaskState::running;
         }
     }//if myidAll==0
         
 
     //Рассылаем количество стартующих задач в данном кванте на все машины
 #ifdef USE_MPI
     MPI_Bcast(&numberOfTask.prepared, 1, MPI_INT, 0, MPI_COMM_WORLD);
 #endif  
     if (myidAll > 0)
     {
         prepList.resize(numberOfTask.prepared);
     }//if myid==0
 
     //следующий фрагмент выполняется только если в данном кванте стартуют новые задачи
     if (numberOfTask.prepared > 0)
     {
         //пересылаем список головных машин стартующих задач на все процессоры
 #ifdef USE_MPI
         MPI_Bcast(prepList.data(), numberOfTask.prepared, MPI_INT, 0, MPI_COMM_WORLD);
 #endif
         
         //формируем группу и коммуникатор стартующих головных процессоров
 #ifdef USE_MPI      
         commStarting = MPI_COMM_NULL;
         MPI_Group_incl(groupAll, numberOfTask.prepared, prepList.data(), &groupStarting);
         MPI_Comm_create(MPI_COMM_WORLD, groupStarting, &commStarting);
 #else
         commStarting = 0x04000000;
         if (numberOfTask.prepared > 0)
             commStarting = 0;
 #endif      
 
         //следующий фрагмент кода только для стартующих процессов, 
         //которые входят в коммуникатор comm_starting
 #ifdef USE_MPI      
         if (commStarting != MPI_COMM_NULL)
 #else
         if (commStarting != 0x04000000)
 #endif
         {
             //получаем номер процесса в списке стартующих
             int myidStarting;
             
 #ifdef USE_MPI          
             MPI_Comm_rank(commStarting, &myidStarting);
 #else
             myidStarting = 0;
 #endif
             
             //подготовка стартующих задач
             parallel.myidWork = 0;
                         
 #ifdef CODE2D           
                 world.reset(new VM2D::World2D(task[myProcState].getPassport()));
 #endif
 
 #ifdef CODE3D                       
                 world.reset(new VM3D::World3D(task[myProcState].getPassport()));
 #endif          
 
             //Коммуникатор головных процессоров стартующих задач
             //выполнил свое черное дело и будет удален
 #ifdef USE_MPI  
             MPI_Comm_free(&commStarting);
 #else
             commStarting = 0x04000000;
 #endif
         } //if(comm_starting != MPI_COMM_NULL)
 
         //их группа тоже больше без надобности
 #ifdef USE_MPI  
         MPI_Group_free(&groupStarting);
 #endif
     }//if (task_prepared>0)
 
     //формируем группу и коммуникатор считающих головных процессоров
     //в том числе тех, которые стартуют, и тех, которые ранее входили в commStarting
     std::vector<int> solvList; //список номеров головных процессов
 
     if (myidAll == 0)
     {
         //Если нулевой процессор свободен 
         //(не является головным в решении задачи в данном кванте) - 
         //- формально присоединяем его в группу головных
         //необходимо для корректного обмена данными
 #ifdef USE_MPI          
         if (procState[0] == MPI_UNDEFINED)
 #else
         if (procState[0] == -32766)
 #endif
             solvList.push_back(0);
 
         //далее ищем считающие задачи (в том числе только что стартующие)
         //и собираем номера их головных процессоров
         //(собираем в порядке просмотра номеров процессоров))
         for (int s = 0; s<nProcAll; ++s)
 #ifdef USE_MPI          
         if ((procState[s] != MPI_UNDEFINED) && (task[procState[s]].state == TaskState::running) && (task[procState[s]].proc[0] == s))
 #else
         if ((procState[s] != -32766) && (task[procState[s]].state == TaskState::running) && (task[procState[s]].proc[0] == s))
 #endif      
             solvList.push_back(s);
 
         sizeCommSolving = static_cast<int>(solvList.size());
 
         flagFinish.clear();
         flagFinish.resize(solvList.size());
 
     }//if myidAll==0
 
     //число sizeCommSolving на единицу больше, чем taskSolving, если нулевой процессор свободен
     //и равно taskSolving, если он занят решением задачи
     //(если он занят решением задачи, то непременно является головным)
 
     //пересылаем число головных процессоров и их список на все компьютеры
     //(в том числе головной процессор - независимо от его состояния)
 #ifdef USE_MPI  
     MPI_Bcast(&sizeCommSolving, 1, MPI_INT, 0, MPI_COMM_WORLD);
 #endif
 
     if (myidAll > 0) 
         solvList.resize(sizeCommSolving);
     
 #ifdef USE_MPI  
     MPI_Bcast(solvList.data(), sizeCommSolving, MPI_INT, 0, MPI_COMM_WORLD);
 #endif
 
 
 #ifdef USE_MPI      
     commSolving = MPI_COMM_NULL;
     
     //формируем группу и коммуникатор головных процессоров
     MPI_Group_incl(groupAll, sizeCommSolving, solvList.data(), &groupSolving);
     MPI_Comm_create(MPI_COMM_WORLD, groupSolving, &commSolving);
 #else
     commSolving = 0x04000000;
     if (sizeCommSolving > 0)
         commSolving = 1;
 #endif
 
     if (myidAll == 0)
     {
         //составление неубывающего массива номеров для верного
         //распределения задач по процессоров, с тем чтобы задачи
         //с прошлого кванта оказались в тех же группах, что и раньше
         ConstructProcStateVar();    
     }
         
     //Пересылаем информацию о состоянии процессоров на все компьютеры
 #ifdef USE_MPI  
     MPI_Scatter(procStateVar.data(), 1, MPI_INT, &myProcStateVar, 1, MPI_INT, 0, MPI_COMM_WORLD);
 #else
     myProcStateVar = procStateVar[0];   
 #endif
 
     //Расщепляем весь коммуникатор MPI_COMM_WORLD (т.е. вообще все процессоры) 
     //на коммуникаторы решаемых задач
     //В результате все процессоры, решающие конкретную задачу,
     //объединяются в коммуникатор comm_work
     //Несмотря на то, что он называется везде одинаково, у каждой задачи он свой
     //и объединяет ровно те процессоры, которые надо
     //Процессоры, состояние которых установлено в "свободен", т.е.
     //ProcState=MPI_UNDEFINED (=-32766) ни в один 
     //новый коммуникатор commWork не попадут
 #ifdef USE_MPI  
     MPI_Comm_split(MPI_COMM_WORLD, myProcStateVar, 0, &parallel.commWork);
 #else
     if (myProcStateVar != -32766)
         parallel.commWork = 0;
     else
         parallel.commWork = 0x04000000;
 #endif
 
     //следующий код выполняется процессорами, участвующими в решении задач, 
     //следовательно, входящими в коммуникаторы comm_work    
 #ifdef USE_MPI  
     if (parallel.commWork != MPI_COMM_NULL)
 #else
     if (parallel.commWork != 0x04000000)
 #endif
     {
         //опеределяем "локальный" номер данного процессора в коммуникаторе commWork и число процессов в нем
 #ifdef USE_MPI  
         MPI_Comm_size(parallel.commWork, &parallel.nProcWork);
         MPI_Comm_rank(parallel.commWork, &parallel.myidWork);
 #else
         parallel.nProcWork = 1;
         parallel.myidWork = 0;
 #endif
         //World равен nullptr только в случае, когда выполняется первый шаг
         //в этом случае он создан только на главном процессоре группы;
         //на остальных происходит его создание
 #ifdef CODE2D
         if (world == nullptr)
             world.reset(new VM2D::World2D(task[myProcState].getPassport()));
 #endif
 
 #ifdef CODE3D
         if (world == nullptr)
             world.reset(new VM3D::World3D(task[myProcState].getPassport()));
 #endif
         
 
         //Синхронизация параметров
 #ifdef USE_MPI
         MPI_Bcast(&(world->getPassportGen().timeDiscretizationProperties.currTime), 1, MPI_DOUBLE, 0, parallel.commWork);
 #endif
         
         if ((parallel.myidWork == 0) && (world->getCurrentStep() == 0))
         {
 #ifdef CODE2D                   
             VM2D::World2D& world2D = dynamic_cast<VM2D::World2D&>(*world);
             //Создание файлов для записи сил
             for (size_t q = 0; q < world2D.getPassport().airfoilParams.size(); ++q)
                 world2D.GenerateMechanicsHeader(q);
             //Создание файла для записи временной статистики
             world2D.getTimestat().GenerateStatHeader();
 #endif
 
 #ifdef CODE3D           
             VM3D::World3D& world3D = dynamic_cast<VM3D::World3D&>(*world);
             //Создание файлов для записи сил
             //for (size_t q = 0; q < world3D.getPassport().airfoilParams.size(); ++q)
             // world3D.GenerateMechanicsHeader(q);
             
             //Создание файла для записи временной статистики
             world3D.getTimestat().GenerateStatHeader();
 #endif
         }
     }
 
 }//TaskSplit()

Here is the call graph for this function:

Here is the caller graph for this function:

void Queue::TaskUpdate ( )

Процедура обновления состояния задач и процессоров

Выполняется в конце очередного кванта

Definition at line 492 of file Queue.cpp.

 {
     info('i') << "------ Kvant finished ------" << std::endl;
 
     //Код только для головных процессов, которые входят в коммуникатор commSolving
     //т.е. выполняется только на головных процессорах, решающих задачи.
     //Кажется, что можно было бы с тем же эфектом написать здесь if (myidWork==0),
     //но это не так, поскольку нужно включить в себя еще нулевой процессор,
     //который в любом случае присоединен к коммуникатору comm_solving,
     //даже если он не решает задачу (а если решает - то он всегда головной)
 #ifdef USE_MPI
     if (commSolving != MPI_COMM_NULL)
     {
         //определяем номер процессора в данном коммуникаторе - 
         // - вдруг потребуется на будущее!
         int myidSolving;
         MPI_Comm_rank(commSolving, &myidSolving);
 
         //Алгоритм возвращения результатов расчета конкретной задачи        
         //Если срабатывает признак того, что решение задачи можно прекращать
         //или не решается ни одна задача (а в комуникатор входит
         //только 0-й процессор, присоединеннй туда насильно)
         int stopSignal = ((parallel.commWork == MPI_COMM_NULL) || (world->isFinished())) ? 1 : 0;
 
         //пересылаем признаки прекращения решения задач на нулевой процессор 
         //коммуникатора comm_solving - т.е. на глобальный процессор с номером 0 ---
         // --- вот для чего его насильно присоединяли к этому коммуникатору
         //если сам по себе он туда не входил        
         MPI_Gather(&stopSignal, 1, MPI_INT, flagFinish.data(), 1, MPI_INT, 0, commSolving);
 
         //после пересылки информации коммуникатор comm_solving уничтожается
         MPI_Comm_free(&commSolving);
 
     } //if(comm_solving != MPI_COMM_NULL)
 
     //и соответствующая ему группа тоже удаляется
     MPI_Group_free(&groupSolving);
 
 
     if (parallel.commWork != MPI_COMM_NULL)
     {
         //К этому месту все процессоры, решающие задачи, приходят уже выполнив все расчеты
         //в рамках одного кванта времени
         //Поэтому коммуникаторы решаемых задач нам уже ни к чему - уничтожаем их
         MPI_Comm_free(&parallel.commWork);
     }
 #else
     if (commSolving != 0x04000000)
     {       
         int myidSolving = 0;    
         int stopSignal = ((parallel.commWork == 0x04000000) || (world->isFinished())) ? 1 : 0;
         flagFinish[0] = stopSignal; 
         commSolving = 0x04000000;
     } //if(comm_solving != 0x04000000)
 
     if (parallel.commWork != 0x04000000)
         parallel.commWork = 0x04000000;
 #endif
 
     //Обновление состояния задач и высвобождение процессоров из отсчитавших задач
     //(выполняется на глобальном нулевом процессоре)
     if (myidAll == 0)
     {
         //Список номеров задач, которые сейчас считаются
         std::vector<int> taskList;
 
         //если состояние "считается" запоминаем эту задачу
         //задачи запоминаются в порядке просмотра процессоров
         for (int s = 0; s<nProcAll; ++s)
 #ifdef USE_MPI
         if ( (procState[s] != MPI_UNDEFINED) && (task[procState[s]].state == TaskState::running) && (task[procState[s]].proc[0] == s) )
 #else
         if ( (procState[s] != -32766) && (task[procState[s]].state == TaskState::running) && (task[procState[s]].proc[0] == s) )
 #endif
             taskList.push_back(procState[s]);
             
         //Тем задачам, которые во флаге прекращения счета вернули "1", 
         //ставим состояние завершения счета.
         //Конструкция +sizeCommSolving-taskSolving введена для смещения в массиве на единицу
         //если нулевой процессор свободен - тогда он присоединяется формально в нулевом
         //элементе массива flagFinish, а содержательная часть массива смещается на 1
         //(в этом случае sizeCommSolving как раз будет на 1 больше, чем taskSolving, см. выше)
         //если же нулевой процесс занят решением задачи, то смещения нет,
         //поскольку в этом случае sizeCommSolving и taskSolving равны
         for (int i = 0; i < numberOfTask.solving; ++i)
         if (flagFinish[i + sizeCommSolving - numberOfTask.solving] == 1)
             task[taskList[i]].state = TaskState::finishing;
     } //if myid==0
 
     if (myidAll == 0)
     {
         for (size_t i = 0; i < task.size(); ++i)
         {
             //освобождаем процессоры от сосчитавшихся задач
             if (task[i].state == TaskState::finishing)
             {
                 //устанавливаем состояние задачи в "отсчитано"
                 task[i].state = TaskState::done;
 
                 //отмечаем последний квант, когда задача считалась
                 task[i].startEndKvant.second = currentKvant;
                 
                 //изменяем счетчики
                 numberOfTask.solving--;
                 numberOfTask.finished++;
 
                 //состояние процессоров, решавших данную задачу, устанавливаем в "свободен"
                 for (int p = 0; p < task[i].nProc; ++p)
                 {
 #ifdef USE_MPI
                     procState[task[i].proc[p]] = MPI_UNDEFINED;
 #else
                     procState[task[i].proc[p]] = -32766;
 #endif
                 }//for p
             }//if (Task[i].state == 3)
         }//for i
 
         //определяем, надо ли еще выделять квант времени или можно заканчивать работу
         //путем сравнения числа отсчитанных задач с общим числом задач
         nextKvant = (numberOfTask.finished < static_cast<int>(task.size())) ? 1 : 0;
     }//if (myid == 0)
 
 #ifdef USE_MPI
     MPI_Bcast(&nextKvant, 1, MPI_INT, 0, MPI_COMM_WORLD);
 #endif
 }//TaskUpdate()

Here is the caller graph for this function:

Member Data Documentation

int VMlib::Queue::commSolving

private

Definition at line 157 of file Queue.h.

int VMlib::Queue::commStarting

private

Definition at line 155 of file Queue.h.

int VMlib::Queue::currentKvant

Номер текущего кванта времени

Definition at line 210 of file Queue.h.

int VMlib::Queue::finished

Число уже решенных задач

Definition at line 95 of file Queue.h.

std::vector<int> VMlib::Queue::flagFinish

private

Список возвращаемых флагов останова задачи

Позиции 0, 1, 2 и т.д. заполняется головными процессами, решающими задачи

Warning: Нулевая позиция — всегда соответствует процессору с myid=0 (глобальному) даже если он не участвует в данном кванте в решении задачи; в этом случае головные процессы заполняют позиции 1, 2, 3 и т.д.

Definition at line 125 of file Queue.h.

int VMlib::Queue::groupAll

private

Definition at line 153 of file Queue.h.

int VMlib::Queue::groupSolving

private

Definition at line 156 of file Queue.h.

int VMlib::Queue::groupStarting

private

Definition at line 154 of file Queue.h.

LogStream VMlib::Queue::info

private

Поток для вывода логов и сообщений об ошибках

Definition at line 177 of file Queue.h.

const double VMlib::Queue::kvantTime = 10.0

Продолжительность кванта времени в секундах

В пределах кванта по времени не производится глобальной синхронизации всех процессоров, решающих все задачи. Проверка состояния очереди и ее обновление производятся между квантами. По исчерпании кванта времени выполнение шагов по времени всеми процессорами прекращается до обновления очереди

Definition at line 207 of file Queue.h.

int VMlib::Queue::myidAll

private

Глобальный номер процессора

Definition at line 128 of file Queue.h.

int VMlib::Queue::myProcState

Состояние данного процессора

Принимает значения :

MPI_UNDEFINED (=-32766, но в зависимости от реализации MPI может быть и другой константой) — свободен;
xxx — занят задачей номер xxx (номер из списка задач).

Definition at line 193 of file Queue.h.

int VMlib::Queue::myProcStateVar

Состояние данного процессора

Принимает значения :

MPI_UNDEFINED (=-32766, но в зависимости от реализации MPI может быть и другой константой) — свободен;
xxx — занят задачей номер xxx (номер из списка задач).

Definition at line 200 of file Queue.h.

int VMlib::Queue::nextKvant

Признак необходимости выполнения следующего кванта и продолжения расчета

Принимает значения:

1 — делать еще один квант (еще есть что считать),
0 — расчет окончен, выход (все расчеты выполнены).

Definition at line 217 of file Queue.h.

int VMlib::Queue::nProcAll

private

Общее число процессоров

Definition at line 131 of file Queue.h.

struct { ... } VMlib::Queue::numberOfTask

Структура, содержащая информацию о количестве задач в данный момент времени

Содержит следующие поля:

prepared — число подготовленных к запуску задач;
solving — число решаемых в данный момент задач;
finished — число уже решенных задач.

Parallel VMlib::Queue::parallel

Класс, опеделяющий параметры исполнения задачи в параллельном MPI-режиме

Внутри него содержатся такие параметры, как:

commWork — коммуникатор для решения конкретной задачи;
myidWork — локальный номер данного процессора в коммуникаторе процессоров, решающих конкретную задачу;
nProcWork — число процессоров, решающих конкретную задачу.

Definition at line 225 of file Queue.h.

int VMlib::Queue::prepared

Число подготовленных к запуску задач

Включено в число решаемых в данный момент задач

Definition at line 87 of file Queue.h.

std::vector<int> VMlib::Queue::procState

private

Список состояний процессоров

Длина списка равна числу процессоров
Заполняется только для на главном узле, у которого myid=0
Принимает значения:

MPI_UNDEFINED (=-32766, но в зависимости от реализации MPI может быть и другой константой) — свободен;
xxx — занят задачей номер xxx (номер из списка задач).

Definition at line 105 of file Queue.h.

std::vector<int> VMlib::Queue::procStateVar

private

Модифицированный список состояний процессоров

Длина списка равна числу процессоров
Заполняется только на главном узле, у которого myid=0
Принимает значения:

MPI_UNDEFINED (=-32766, но в зависимости от реализации MPI может быть и другой константой) — свободен;
xxx — занят задачей номер xxx (номер из списка задач).
Аналогичен procState, но отличается от него следующим:
процессы, занятые одной задачей, получают один и тот же номер;
позиции, в которых впервые появляются номера в этом списке, образуют возрастающую последовательность.

Definition at line 117 of file Queue.h.

int VMlib::Queue::sizeCommSolving

private

Число процессоров в группе для головных процессоров в решаемых в данном кванте времени задачах

В него всегда входит 0-й процессор, даже если он не раешает задачу в данном кванте времени

Definition at line 163 of file Queue.h.

int VMlib::Queue::solving

Число решаемых в данный момент задач

Включает число подготовленных к запуску задач

Definition at line 92 of file Queue.h.

std::vector<Task> VMlib::Queue::task

Список описаний решаемых задач

В описаниях содержится информация о прохождении задач и их текущем состоянии

Definition at line 183 of file Queue.h.

std::unique_ptr<WorldGen> VMlib::Queue::world

Умный указатель на текущую решаемую задачу

Definition at line 186 of file Queue.h.

The documentation for this class was generated from the following files:

VMlib/Queue/Queue.h
VMlib/Queue/Queue.cpp

Public Member Functions

Public Attributes

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation