From 5d8ed721465b1e84eff6b5be1e5c757ca2023bc0 Mon Sep 17 00:00:00 2001 From: Ovidiu Mircescu Date: Thu, 4 Jun 2020 17:53:31 +0200 Subject: [PATCH] Workload manager: source code organization. --- src/engine/CMakeLists.txt | 2 + src/engine/Executor.cxx | 87 ++--------------------------------- src/engine/WlmTask.cxx | 96 +++++++++++++++++++++++++++++++++++++++ src/engine/WlmTask.hxx | 47 +++++++++++++++++++ 4 files changed, 148 insertions(+), 84 deletions(-) create mode 100644 src/engine/WlmTask.cxx create mode 100644 src/engine/WlmTask.hxx diff --git a/src/engine/CMakeLists.txt b/src/engine/CMakeLists.txt index d8f4d296d..b4d9f7bfd 100644 --- a/src/engine/CMakeLists.txt +++ b/src/engine/CMakeLists.txt @@ -116,6 +116,7 @@ SET(YACSlibEngine_HEADERS PlayGround.hxx ObserverAsPlugin.hxx NbBranches.hxx + WlmTask.hxx ) # --- sources --- @@ -194,6 +195,7 @@ SET(YACSlibEngine_SOURCES PlayGround.cxx ComplexWeight.cxx ObserverAsPlugin.cxx + WlmTask.cxx ) SET(YACSlibEngine_HEADERS ${YACSlibEngine_HEADERS} PARENT_SCOPE) # Make it visible to src/engine_swig to handle dependencies diff --git a/src/engine/Executor.cxx b/src/engine/Executor.cxx index 24603ef0a..d899ae995 100644 --- a/src/engine/Executor.cxx +++ b/src/engine/Executor.cxx @@ -30,6 +30,7 @@ #include "ServiceNode.hxx" #include "ComposedNode.hxx" +#include "WlmTask.hxx" #include "workloadmanager/WorkloadManager.hxx" #include "workloadmanager/DefaultAlgorithm.hxx" @@ -1664,88 +1665,6 @@ void Executor::makeDatastreamConnections(Task *task) traceExec(task, "state:"+Node::getStateName(task->getState()),ComputePlacement(task)); } -#include "Runtime.hxx" -static -void loadResources(WorkloadManager::WorkloadManager& wm) -{ - Runtime *r(getRuntime()); - if(!r) - throw YACS::Exception("loadResources : no runtime !"); - std::vector< std::pair > data(r->getCatalogOfComputeNodes()); - int id = 0; - for(const std::pair& res : data) - { - WorkloadManager::Resource newResource; - newResource.name = res.first; - newResource.id = id; - id++; - newResource.nbCores = res.second; - wm.addResource(newResource); - } -} - -class NewTask : public WorkloadManager::Task -{ -public: - NewTask(Executor& executor, YACS::ENGINE::Task* yacsTask); - const WorkloadManager::ContainerType& type()const override; - void run(const WorkloadManager::RunInfo& runInfo)override; - bool isAccepted(const WorkloadManager::Resource& r)override; -private: - WorkloadManager::ContainerType _type; - Executor& _executor; - YACS::ENGINE::Task * _yacsTask; -}; - -NewTask::NewTask(Executor& executor, YACS::ENGINE::Task* yacsTask) -: _type() -, _executor(executor) -, _yacsTask(yacsTask) -{ - Container * yacsContainer = yacsTask->getContainer(); - if(yacsContainer != nullptr && yacsTask->canAcceptImposedResource()) - { - _type.ignoreResources = false; - _type.name = yacsContainer->getName(); - std::string nb_procs_str = yacsContainer->getProperty("nb_parallel_procs"); - float needed_cores = 0.0; - if(!nb_procs_str.empty()) - needed_cores = std::stof(nb_procs_str); - _type.neededCores = needed_cores; - } - else - { - _type.ignoreResources = true; - _type.name = "test"; - _type.neededCores = 0; - } - _type.id = 0; -} - -const WorkloadManager::ContainerType& NewTask::type()const -{ - return _type; -} - -void NewTask::run(const WorkloadManager::RunInfo& runInfo) -{ - _executor.loadTask(_yacsTask, runInfo); - _executor.makeDatastreamConnections(_yacsTask); - YACS::Event ev = _executor.runTask(_yacsTask); - _executor.endTask(_yacsTask, ev); - delete this; // provisoire -} - -bool NewTask::isAccepted(const WorkloadManager::Resource& r) -{ - Container * yacsContainer = _yacsTask->getContainer(); - std::string hostname = yacsContainer->getProperty("hostname"); - bool accept = true; - if(!hostname.empty()) - accept = (hostname == r.name); - return accept; -} - void Executor::runWlm(Scheduler *graph,int debug, bool fromScratch) { DEBTRACE("Executor::runWlm debug: "<< graph->getName() <<" "<< debug<<" fromScratch: "<getContainer(); + if(yacsContainer != nullptr && yacsTask->canAcceptImposedResource()) + { + _type.ignoreResources = false; + _type.name = yacsContainer->getName(); + std::string nb_procs_str = yacsContainer->getProperty("nb_parallel_procs"); + float needed_cores = 0.0; + if(!nb_procs_str.empty()) + needed_cores = std::stof(nb_procs_str); + _type.neededCores = needed_cores; + } + else + { + _type.ignoreResources = true; + _type.name = "test"; + _type.neededCores = 0; + } + _type.id = 0; +} + +const WorkloadManager::ContainerType& WlmTask::type()const +{ + return _type; +} + +void WlmTask::run(const WorkloadManager::RunInfo& runInfo) +{ + _executor.loadTask(_yacsTask, runInfo); + _executor.makeDatastreamConnections(_yacsTask); + YACS::Event ev = _executor.runTask(_yacsTask); + _executor.endTask(_yacsTask, ev); + delete this; // provisoire +} + +bool WlmTask::isAccepted(const WorkloadManager::Resource& r) +{ + Container * yacsContainer = _yacsTask->getContainer(); + std::string hostname = yacsContainer->getProperty("hostname"); + bool accept = true; + if(!hostname.empty()) + accept = (hostname == r.name); + return accept; +} + +void WlmTask::loadResources(WorkloadManager::WorkloadManager& wm) +{ + Runtime *r(getRuntime()); + if(!r) + throw YACS::Exception("loadResources : no runtime !"); + std::vector< std::pair > data(r->getCatalogOfComputeNodes()); + int id = 0; + for(const std::pair& res : data) + { + WorkloadManager::Resource newResource; + newResource.name = res.first; + newResource.id = id; + id++; + newResource.nbCores = res.second; + wm.addResource(newResource); + } +} + +} +} diff --git a/src/engine/WlmTask.hxx b/src/engine/WlmTask.hxx new file mode 100644 index 000000000..d97889d5d --- /dev/null +++ b/src/engine/WlmTask.hxx @@ -0,0 +1,47 @@ +// Copyright (C) 2006-2020 CEA/DEN, EDF R&D +// +// This library is free software; you can redistribute it and/or +// modify it under the terms of the GNU Lesser General Public +// License as published by the Free Software Foundation; either +// version 2.1 of the License, or (at your option) any later version. +// +// This library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +// Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public +// License along with this library; if not, write to the Free Software +// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA +// +// See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com +// +#ifndef __WLM_TASK_HXX__ +#define __WLM_TASK_HXX__ +#include "Executor.hxx" +#include "Task.hxx" +#include "workloadmanager/WorkloadManager.hxx" + +namespace YACS +{ +namespace ENGINE +{ +class YACSLIBENGINE_EXPORT WlmTask : public WorkloadManager::Task +{ +public: + WlmTask(Executor& executor, YACS::ENGINE::Task* yacsTask); + const WorkloadManager::ContainerType& type()const override; + void run(const WorkloadManager::RunInfo& runInfo)override; + bool isAccepted(const WorkloadManager::Resource& r)override; + + static void loadResources(WorkloadManager::WorkloadManager& wm); +private: + WorkloadManager::ContainerType _type; + Executor& _executor; + YACS::ENGINE::Task * _yacsTask; +}; + +} +} + +#endif //__WLM_TASK_HXX__ -- 2.39.2