VMlib::Queue Class Reference

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

#include <Queue.h>

Collaboration diagram for VMlib::Queue:

Public Member Functions
	Queue (int &argc, char **&argv)
	Конструктор
	~Queue ()
	Деструктор
void	TaskSplit ()
	Процедура постановка новых задач на отсчет и занятие процессоров
void	TaskUpdate ()
	Процедура обновления состояния задач и процессоров
void	RunConveyer ()
	Запуск вычислительного конвейера (в рамках кванта времени)
void	LoadTasksList (const std::string &_tasksFile, const std::string &_mechanicsFile, const std::string &_defaultsFile, const std::string &_switchersFile)
	Загрузка списка задач

Public Attributes
std::vector< Task >	task
	Список описаний решаемых задач
std::unique_ptr< WorldGen >	world
	Умный указатель на текущую решаемую задачу
int	myProcState
	Состояние данного процессора
int	myProcStateVar
	Состояние данного процессора
const double	kvantTime = 1.0
	Продолжительность кванта времени в секундах
int	currentKvant
	Номер текущего кванта времени
int	nextKvant
	Признак необходимости выполнения следующего кванта и продолжения расчета
Parallel	parallel
	Класс, опеделяющий параметры исполнения задачи в параллельном MPI-режиме

Private Member Functions
void	ConstructProcStateVar ()
	Процедура, нумерующая задачи в возрастающем порядке
void	AddTask (int _nProc, std::unique_ptr< PassportGen > _passport)
	Добавление задачи в список

Private Attributes
struct {
int   prepared
	Число подготовленных к запуску задач More...
int   solving
	Число решаемых в данный момент задач More...
int   finished
	Число уже решенных задач More...
}	numberOfTask
	Структура, содержащая информацию о количестве задач в данный момент времени
std::vector< int >	procState
	Список состояний процессоров
std::vector< int >	procStateVar
	Модифицированный список состояний процессоров
std::vector< int >	flagFinish
	Список возвращаемых флагов останова задачи
int	myidAll
	Глобальный номер процессора
int	nProcAll
	Общее число процессоров
int	groupAll
int	groupStarting
int	commStarting
int	groupSolving
int	commSolving
int	sizeCommSolving
	Число процессоров в группе для головных процессоров в решаемых в данном кванте времени задачах
LogStream	info
	Поток для вывода логов и сообщений об ошибках

Detailed Description

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

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

Author

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

Version

1.14

Date

6 марта 2026 г.

Definition at line 75 of file Queue.h.

Constructor & Destructor Documentation

Queue()

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

Конструктор

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

Parameters

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

Definition at line 78 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);      
#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::~Queue

(

)

Деструктор

Definition at line 119 of file Queue.cpp.

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

Member Function Documentation

AddTask()

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

private

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

Parameters

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

Definition at line 702 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:

ConstructProcStateVar()

void Queue::ConstructProcStateVar ( )

private

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

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

Definition at line 616 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:

LoadTasksList()

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 713 of file Queue.cpp.

{
    std::string extTasksFile = _tasksFile;
    std::string extMechanicsFile = _mechanicsFile;
    std::string extDefaultsFile = _defaultsFile;
    std::string extSwitchersFile = _switchersFile;
 
    if (
        fileExistTest(extTasksFile, info, true, { "txt", "TXT" }) &&
        fileExistTest(extDefaultsFile, info, true, { "txt", "TXT" }) &&
        fileExistTest(extSwitchersFile, info, true, { "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;
                    exit(-1);
                }
 
                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::CreateUserDirectory(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, "./" + dir + "/" + 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:

RunConveyer()

void Queue::RunConveyer

(

)

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

Definition at line 656 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:

TaskSplit()

void Queue::TaskSplit

(

)

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

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

Definition at line 129 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 << "] <=> " << ((procState[i] >= 0) ? std::string("problem[" + std::to_string(procState[i]) + "]") : std::string("free")) << 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
        
        
        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.getTimers().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);
#endif
        }
    }
 
}//TaskSplit()

Here is the call graph for this function:

Here is the caller graph for this function:

TaskUpdate()

void Queue::TaskUpdate

(

)

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

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

Definition at line 486 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

commSolving

int VMlib::Queue::commSolving

private

Definition at line 160 of file Queue.h.

commStarting

int VMlib::Queue::commStarting

private

Definition at line 158 of file Queue.h.

currentKvant

int VMlib::Queue::currentKvant

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

Definition at line 213 of file Queue.h.

finished

int VMlib::Queue::finished

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

Definition at line 98 of file Queue.h.

flagFinish

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

private

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

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

Warning

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

Definition at line 128 of file Queue.h.

groupAll

int VMlib::Queue::groupAll

private

Definition at line 156 of file Queue.h.

groupSolving

int VMlib::Queue::groupSolving

private

Definition at line 159 of file Queue.h.

groupStarting

int VMlib::Queue::groupStarting

private

Definition at line 157 of file Queue.h.

info

LogStream VMlib::Queue::info

private

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

Definition at line 180 of file Queue.h.

kvantTime

const double VMlib::Queue::kvantTime = 1.0

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

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

Definition at line 210 of file Queue.h.

myidAll

int VMlib::Queue::myidAll

private

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

Definition at line 131 of file Queue.h.

myProcState

int VMlib::Queue::myProcState

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

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

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

Definition at line 196 of file Queue.h.

myProcStateVar

int VMlib::Queue::myProcStateVar

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

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

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

Definition at line 203 of file Queue.h.

nextKvant

int VMlib::Queue::nextKvant

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

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

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

Definition at line 220 of file Queue.h.

nProcAll

int VMlib::Queue::nProcAll

private

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

Definition at line 134 of file Queue.h.

[struct]

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

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

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

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

parallel

Parallel VMlib::Queue::parallel

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

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

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

Definition at line 228 of file Queue.h.

prepared

int VMlib::Queue::prepared

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

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

Definition at line 90 of file Queue.h.

procState

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

private

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

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

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

Definition at line 108 of file Queue.h.

procStateVar

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

private

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

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

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

Definition at line 120 of file Queue.h.

sizeCommSolving

int VMlib::Queue::sizeCommSolving

private

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

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

Definition at line 166 of file Queue.h.

solving

int VMlib::Queue::solving

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

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

Definition at line 95 of file Queue.h.

task

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

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

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

Definition at line 186 of file Queue.h.

world

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

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

Definition at line 189 of file Queue.h.

---

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

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