VisualSapfor/Planner/Supervisor.h

#pragma once

#include <map>
#include <set>
#include <vector>
#include <queue>
#include "File.h"   
#include "Task.h"   

enum SupervisorState {   
    WorkspacesCreation, //0   
    Preparation,		//1   
    Execution,		//2   
    Archivation,        //3
    End			//4   
};    
  
template <class T>     
class Supervisor : public Array <T> {
protected:
	SupervisorState state;
public:
	virtual String getStatePrefix() {
		return String("");
	}
	String printState() {
		switch (state) {
		case WorkspacesCreation:
			return String("WorkspacesCreation");
		case Preparation:
			return String("Preparation");
		case Execution:
			return String("Execution");
		case Archivation:
			return String("Archivation");
		case End:
			return String("End");
		default:
			return "?";
		}
	}
	//-
	void print() {
		for (auto& elem : this->getElements())
			elem->print();
	}

	void init(const char* fileName, int recordSize) {
		state = WorkspacesCreation;
		File* packedTasks = new File(fileName);
		Text* lines = packedTasks->readLines();

		const long length = lines->getLength() / recordSize;
		int offset = 0;
		for (int i = 0; i < length; ++i) {
			this->add(new T(lines, offset));
			offset += recordSize;
		}
		delete packedTasks;
		delete lines;
	}

	void changeState() {
		switch (this->state) {
		case WorkspacesCreation:
			this->state = Preparation;
			saveState();
			break;
		case Preparation:
			this->state = Execution;
			saveState();
			break;
		case Execution:
			Finalize();
			this->state = End;
			saveState();
			break;
		default:
			this->state = End;
			break;
		}
	}

	void Do() {
		saveState();
		long activeCount = 0;
		//todo обязательно убрать отладочную печать.  
		printf("tasks count = %ld\n", this->getLength());
		while (this->state != End) {
			//			printf("state=%d\n", this->state);  
			//			printf("max=%d; busy=%d; free=%d\n", maxKernels, busyKernels, freeKernels);			   
			activeCount = 0;
			for (long i = 0; i < this->getLength(); ++i) {
				T* task = this->get(i);
				switch (this->state) {
				case WorkspacesCreation:
					if (task->getState() == Waiting) {
						activeCount++;
						task->createWorkspace();
						task->setState(WorkspaceCreated);
					}
					break;
				case Preparation:
					if (task->getState() == WorkspaceCreated) {
						activeCount++;
						task->prepareWorkspace();
						task->createLaunchScript();
						task->setState(WorkspaceReady);
					}
					break;
				case Execution:
					if (task->getState() == WorkspaceReady) {
						activeCount++;
						task->Start();
					}
					else if (task->getState() == Running) {
						activeCount++;
						task->Check();
					}
					break;
				default:
					//						printf("id = %ld; state = %d\n", task->getId(), task->getState());   
					break;
				}
			}
			//			printf("active count = %d\n", activeCount);   
			if (activeCount == 0)
				changeState();
			Utils::Sleep(2);
		}
	}

	void DoWithSchedule(int maxKernels) {
		saveState();

		// подготовка тестов
		while (this->state != Execution) {
			for (auto& task : this->getElements()) {
				switch (this->state) {
				case WorkspacesCreation:
					if (task->getState() == Waiting) {
						task->createWorkspace();
						task->setState(WorkspaceCreated);
					}
					break;
				case Preparation:
					if (task->getState() == WorkspaceCreated) {
						task->prepareWorkspace();
						task->createLaunchScript();
						task->setState(WorkspaceReady);
					}
					break;
				default:
					//printf("id = %ld; state = %d\n", task->getId(), task->getState());   
					break;
				}
			}
			changeState();
		}

		map<int, queue<T*>, std::greater<int>> sortedByKernelNeeds;

		long activeTasks = 0;
		for (auto& task : this->getElements()) {			
			if (task->getState() == WorkspaceReady) {
				activeTasks++;
				sortedByKernelNeeds[task->getKernels()].push(task);
			}
		}

		printf("total tasks count = %ld, active task count %ld, maxkernels %d\n", this->getLength(), activeTasks, maxKernels);

		int busyKernels = 0;
		set<T*> activeTaskSet;
		bool ignoreCheck = true;

		while (activeTasks) {
			vector<int> emptyKeys;

			//ставим задачи от больших к меньшему по ядрам
			for (auto& elem : sortedByKernelNeeds) {
				int freeKernels = maxKernels - busyKernels;
				int kernelsNeeded = elem.first;
				
				while (kernelsNeeded <= freeKernels && elem.second.size()) {
					T* task = elem.second.front();
					elem.second.pop();

					activeTaskSet.insert(task);
					task->Start(ignoreCheck);
					printf("start task with %d kernels and id %ld\n", task->getKernels(), task->getId());

					busyKernels += task->getKernels();
					freeKernels = maxKernels - busyKernels;
				}

				if (elem.second.size() == 0)
					emptyKeys.push_back(kernelsNeeded);

				//если ядер не осталось, то нет смысла дальше смотреть
				if (freeKernels == 0)
					break;				
			}

			// очищаем от пустых ключей
			for (auto& empty : emptyKeys)
				sortedByKernelNeeds.erase(empty);
				
			// проверяем нет ли завершившихся задач
			for (auto it = activeTaskSet.begin(); it != activeTaskSet.end(); )
			{
				T* task = *(it);
				
				if (task->Check()) {
					it++;
					activeTaskSet.erase(task);
					activeTasks--;
					busyKernels -= task->getKernels();
					printf("  done task with %d kernels and id %ld\n", task->getKernels(), task->getId());

					continue;
				}
				it++;
			}
		}

		changeState();
	}

	virtual void Finalize() {}
	void saveState() {
		Utils::Sleep(1); //чтобы не было одинаковых по дате файлов.
		String stateFile = packageWorkspace + "/state/" + getStatePrefix() + printState();
		//printf("stateFile=<%s>\n", stateFile.getCharArray()); 
		File tmp(stateFile, Utils::getDate());
	}
};