TCluster.h

Go to the documentation of this file.

// Copyright (C) 2013-2024 CEA/DES
//
// This program 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 3 of the License, or any
// later version.
//
// This program 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 program. If not, see <http://www.gnu.org/licenses/>.
// $Id$
// $Author$
// $Date$
// $Revision$
// $State$

#ifndef TCLUSTER_H
#define TCLUSTER_H

#include <stdlib.h>
#include <iostream>

//ROOT
#include "Rtypes.h"
#include "TNamed.h"

// Uranie
#include "Launcher.h"

#ifdef WITH_LIBSSH
#include "TCode.h"
#include "TInputFileFlag.h"
#include "TAttribute.h"
#include "TInputFileRecreate.h"

//SSH
#define LIBSSH_STATIC 1
#include <libssh/libssh.h>
#include <libssh/sftp.h>
#endif

namespace URANIE
{
namespace Launcher
{
class TCluster: public TNamed
{
public:
    Bool_t _blog;                          
#ifdef WITH_LIBSSH    
    enum EDistrib { kUnknown, kLSB, kSLURM, kMPI, kPBS, kSGE, kEc };
    EDistrib  _edistrib;
    TString _eType;
#endif
    
private:
    Int_t _nnode;                          
    TString _slistOfNodes;                   
public:
    //---------------------------------------------

    TCluster(const char* name, const char *nodeList, Int_t nnode);
#ifdef WITH_LIBSSH    
    TCluster(URANIE::DataServer::TDataServer *tds, TCode *code);
    TCluster(TCode *code);
#endif

    virtual ~TCluster();

    virtual void init() = 0;

#ifdef WITH_LIBSSH    

    void setWorkingDirectory(TString localWorkingDirectory);
    void setRemoteWorkingDirectory();

    TCode * _code;
    URANIE::DataServer::TDataServer* _tds;
#endif
    //---------------------------------------------
    void setNNodes(Int_t nnode){
        _nnode = nnode;
    }
    TString getListOfNodes(){
        return _slistOfNodes;
    }
    // Adds the node name
    void addNodeName(const TString& snodename, Int_t n = -1);
    virtual void getCommandPrefix(char *, char*) = 0;

    //---------------------------------------------
    void setLog(){
        _blog = kTRUE;
    }
    void unsetLog(){
        _blog = kFALSE;
    }
    void changeLog(){
        _blog = _blog ? kFALSE : kTRUE;
    }
    Bool_t getLog(){
        return _blog;
    }

    virtual void printLog(Option_t *option = "");

#ifdef WITH_LIBSSH

    void setClusterUserAuth(TString user ,TString  authentification_method="public_key"){
        _sClusterUser=user;
        _sClusterUserAuthMethod=authentification_method;
    }

    virtual void setCluster(TString cluster_name){
        _sClusterName = cluster_name;

    }
    virtual void setCluster(TCluster *cluster);

    TString getClusterName() const{
        return _sClusterName;
    }
    TString getClusterIP() const{
        return _sClusterIP;
    }
    TString setClusterIP() const{
        return _sClusterIP;
    }

    void setSkeleton(TString skeleton){
        _sSkeletonHeader = skeleton;
    }
    TString getSkeleton() const{
        return _sSkeletonHeader;
    }

    void setOutputHeaderName(TString oHeaderName){
#ifdef WIN32
        const char *tail = strrchr(oHeaderName.Data(), '/');
        _sOutputHeader = tail ? tail + 1 : oHeaderName.Data();   
#else
        _sOutputHeader=basename(oHeaderName);
#endif
    }
    //        //! Sets the path to generate header output path
    //    /*!
    //     * \param generate header (TString) output of the cluster
    //     */

    //     * \brief Get the path to remoteWorkingDir
    //     * \return Output Path Header Name
    //     */


    void setJobID(TString jobid){
        _sJobID.push_back(jobid);

    }

    void setNRetries(Int_t retries){
        _sRetries = retries;

    }

    TString getOutputHeaderName() const{
        return _sOutputHeader;
    }

    TString getOutputHeaderPath() const{
        return _sOutputHeaderPath;
    }

    TString getClusterUser() const{
        return _sClusterUser;
    }

    TString getClusterUserAuthMethod() const{
        return _sClusterUserAuthMethod;
    }

    TString getRemotePath() const{
        return _sRemoteWorkingDirectory;
    }

    TString getOutputSsh() const{
        return _sSshOutput;
    }

    TString getSplitName() const{
        return _sSplitName;
    }

    TString getJobID(Int_t index) const{
        return _sJobID.at(index);
    }

    void setNumberOfCores(Int_t ncores){
        _nAllocatedCores = ncores;
    }
    Int_t getNumberOfCores() const{
        return _nAllocatedCores;
    }
    //CLUSTER CODE
    void preTreatment(Option_t *option="");


    void addClusterDirective(TString directive, TString value);


    void setJobName(TString path){
        _sJobName=path;
    }

    TString getJobName(){
        return _sJobName;
    }

    // SSH


    Int_t connectSSH(TString cluster_name ,  TString  user,  TString authentification_method);

    Int_t verify_knownhostSSH();

    Int_t openSFTPChannel();
    Int_t sendHeader();

    Int_t sendCompiledBinary(std::list<TString> filenames);

    Int_t sendInputFiles(Int_t start, Int_t end);


    Int_t retrieveInputFiles(const char * localFolder);


    Int_t closeSFTPChannel();


    void setRemotePath(TString remotePath){
        if (!remotePath.EndsWith("/"))
        {
            remotePath.Append("/");
        }
        _sRemoteWorkingDirectory = remotePath;
    }


    void setDataServer(URANIE::DataServer::TDataServer *tds){
        _tds =tds;
    }

    TString getGateway()
    {
        return _sGateWay;
    }

    Bool_t isConnected(){
        return _sConnected;
    }

    virtual void selectHeader(TString SkeletonHeaderName="./",TString localHeaderFolderPath="" )   {UNUSED(SkeletonHeaderName,localHeaderFolderPath); cout << "base" <<endl;};
    virtual void generateHeader(Option_t *option = "")   {UNUSED(option); cout << "base" <<endl;};
    virtual void generateHeaderSplit(Option_t *option = "")  {UNUSED(option); cout << "base" <<endl;};
    TString getJobStatusCommand(std::list<std::pair<TString, std::pair<int, int> > > JobList) {UNUSED(JobList); return "base";};
    TString getJobStatusCommand(TString JobIDs) {UNUSED(JobIDs); return "base";};
    TString checkJobStatus(TString JobId, TString Command) {UNUSED(JobId,Command); return "base";};
    void  substituteDirectives(TString infilePath, TString outFilePath, std::map<TString,TString> ListOfDirectives);

    void execRemoteCommand(TString command);
    virtual TString submitJob(TString command);

    void replaceHeaderKey(std::string& source, const std::string& from, const std::string& to);
#endif
    
protected:

    // SSH
#ifdef WITH_LIBSSH    
    ssh_session _sSshSession;
    sftp_session _sFtpSession;
    ssh_session *_ptrsshSession;
    sftp_session * _ptrsftp;

    TString _sSshOutput;
    Bool_t _sConnected;

    // Header
    TString _sSkeletonHeader;
    TString _sOutputHeader;
    TString _sSplitName;
    TString _sWorkingDirectory;
    TString _sOutputHeaderPath;
    TString _sJobName;
    TString _slocalHeaderFolderPath;
    Int_t _nAllocatedCores;
    Int_t _sRetries;

    TString _sRemoteWorkingDirectory;

    // Cluster

    TString _sClusterUser;
    TString _sClusterUserAuthMethod;
    TString _sClusterName;
    TString _sClusterIP;

    TString _sGateWay;
    vector<TString> _sJobID;
    Bool_t _sUserBinary=kFALSE;
    TString _sConnectionProtocol="ssh";

    std::map<TString,TString> _sListClusterDirectives;
#endif

    Int_t getProcsPerJob(Char_t separator, const Char_t* procs) const;

    ClassDef(URANIE::Launcher::TCluster, ID_LAUNCHER)

};


} // Fin du namespace Launcher
} // Fin du namespace URANIE
#endif