mwtask.h

Go to the documentation of this file.

 
 
 
// Copyright (C) 1997-2020 by Pawel Pilarczyk.
//
// This file is part of my research software package. This is free software:
// you can redistribute it and/or modify it under the terms of the GNU
// General Public License as published by the Free Software Foundation,
// either version 3 of the License, or (at your option) any later version.
//
// This software 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 General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this software; see the file "license.txt". If not,
// please, see <https://www.gnu.org/licenses/>.
 
// Started on August 11, 2004. Last revision: November 29, 2007.
 
 
#ifndef _CHOMP_MULTIWORK_MWTASK_H_
#define _CHOMP_MULTIWORK_MWTASK_H_
 
#include <fstream>
#include <string>
#include <vector>
#include <cstdlib> // for "atoi"
#include <ctime> // for log start/stop time
 
#include "chomp/multiwork/mwconfig.h"
#include "chomp/multiwork/mwlowlev.h"
#include "chomp/multiwork/mwdata.h"
 
 
namespace chomp {
namespace multiwork {
 
// --------------------------------------------------
// ----------------- message codes ------------------
// --------------------------------------------------
 
enum mwCodes
{
        mwNoMsg = 0x0000,
 
        mwInitMsg = 0x0001,
 
        mwStdMsg = 0x0002,
 
        mwKeepMsg = 0x0004,
 
        mwDontKeepMsg = 0x0008,
 
        mwByeMsg = 0x0010,
 
        mwRejectedMsg = 0x0001,
 
        mwPortMsg = 0x0002
 
}; /* enum mwCodes */
 
 
// --------------------------------------------------
// --------------------- mwTask ---------------------
// --------------------------------------------------
 
class mwTask
{
public:
        mwTask ();
                
        virtual ~mwTask ();
 
        // --- configuration ---
 
        void Port (int number);
 
        int Port () const;
 
        void ControlNumber (unsigned int number);
 
        unsigned int ControlNumber () const;
 
        void TimeOut (int seconds);
 
        int TimeOut () const;
 
        // --- log file ---
 
        int LogFile (const char *filename);
 
        void LogFile (const mwTask &other);
 
        void LogClose ();
 
        // --- computer addresses ---
 
        int Add (const char *name, int port = -1);
        
        int Load (const char *filename);
        
        // --- quit waiting workers ---
 
        int QuitWorkers ();
 
private:
        // --- configuration data ---
        
        int portnum;
 
        unsigned int ctrlnum;
 
        int timeout;
 
        // --- log file ---
 
protected:
        std::ofstream *logFile;
 
private:
        bool logBorrowed;
 
protected:
        // --- network communication ---
 
        static int SendMessage (int fd, unsigned int ctrl,
                unsigned int code, const mwData &x);
 
        static int RecvMessage (int fd, unsigned int &ctrl,
                unsigned int &code, mwData &x);
 
        // --- computer addresses ---
 
        std::vector<std::string> computers;
 
        std::vector<int> ports;
 
private:
        // --- other stuff ---
 
        mwTask (const mwTask &) {}
                
        mwTask &operator = (const mwTask &) {return *this;}
                
}; /* class mwTask */
 
// --------------------------------------------------
 
inline mwTask::mwTask ():
        portnum (mwPORT),
        ctrlnum (mwCTRLNUM),
        timeout (mwTIMEOUT),
        logFile (0),
        logBorrowed (false)
{
        return;
} /* mwTask::mwTask */
 
// --------------------------------------------------
 
inline void mwTask::LogClose ()
{
        if (!logFile)
                return;
        if (logBorrowed)
        {
                logFile = 0;
                logBorrowed = false;
                return;
        }
        std::time_t stop_time;
        std::time (&stop_time);
        *logFile << "\nMultiWork log file closed on " <<
                std::asctime (std::localtime (&stop_time)) << "\n"
                "-----------------------------------------------------\n" <<
                std::endl;
        delete logFile;
        logFile = 0;
        return;
} /* mwTask::LogClose */
 
inline int mwTask::LogFile (const char *filename)
{
        // close the current log if in use
        if (logFile)
                LogClose ();
 
        // if no file name supplied, return now
        if (!filename || !*filename)
                return mwOk;
 
        // create a file stream variable
        logFile = new std::ofstream;
        if (!logFile)
                return mwError;
 
        // open the log file for appending
        logFile -> open (filename, std::ios::app);
 
        // write the current time to the log
        std::time_t start_time;
        std::time (&start_time);
        *logFile << "MultiWork log file opened on " <<
                std::asctime (std::localtime (&start_time)) << std::endl;
 
        // if unable to open the file or an error occurred, return mwError
        if (!*logFile)
        {
                delete logFile;
                return mwError;
        }
 
        return mwOk;
} /* mwTask::LogFile */
 
inline void mwTask::LogFile (const mwTask &other)
{
        // close the current log if in use
        LogClose ();
 
        // if there is no other log file, do nothing
        if (!other. logFile)
                return;
 
        // borrow the log file pointer
        logBorrowed = true;
        logFile = other. logFile;
 
        return;
} /* mwTask::LogFile */
 
// --------------------------------------------------
 
inline mwTask::~mwTask ()
{
        // close the log file if it was in use
        LogClose ();
 
        return;
} /* mwTask::~mwTask */
 
// --------------------------------------------------
 
inline void mwTask::Port (int number)
{
        if (number >= 0)
                portnum = static_cast<short int> (number);
        return;
} /* mwTask::Port */
 
inline int mwTask::Port () const
{
        return portnum;
} /* mwTask::Port */
 
inline void mwTask::ControlNumber (unsigned int number)
{
        ctrlnum = number;
        return;
} /* mwTask::ControlNumber */
 
inline unsigned int mwTask::ControlNumber () const
{
        return ctrlnum;
} /* mwTask::ControlNumber */
 
inline void mwTask::TimeOut (int seconds)
{
        timeout = seconds;
        return;
} /* mwTask::TimeOut */
 
inline int mwTask::TimeOut () const
{
        return timeout;
} /* mwTask::TimeOut */
 
// --------------------------------------------------
 
inline int mwTask::Add (const char *name, int port)
{
        // if the name is empty, ignore it
        if (!name || !*name)
                return mwError;
 
        // determine whether the name contains a colon and port number
        int pos = 1;
        while (name [pos])
                ++ pos;
        -- pos;
        while (pos && (name [pos] != ':') &&
                (name [pos] >= '0') && (name [pos] <= '9'))
        {
                -- pos;
        }
 
        // if the name contains colon and some digits after the colon...
        if (pos && (name [pos] == ':') && name [pos + 1])
        {
                // append the computer name and the chosen port number
                char *compname = new char [pos + 1];
                for (int i = 0; i < pos; ++ i)
                        compname [i] = name [i];
                compname [pos] = '\0';
                port = std::atoi (name + pos + 1);
                if (port <= 0)
                        return mwError;
                computers. push_back (std::string (compname));
                ports. push_back (port);
                delete [] compname;
        }
        else
        {
                // if the port number is not reasonable, use the default one
                if (port <= 0)
                        port = portnum;
                if (port <= 0)
                        return mwError;
                computers. push_back (std::string (name));
                ports. push_back (port);
        }
        return mwOk;
} /* mwTask::Add */
 
inline int mwTask::Load (const char *filename)
{
        std::ifstream f (filename);
        if (!f)
                return mwError;
 
        char buf [512];
        int counter = 0;
        while (1)
        {
                *buf = '\0';
                f. getline (buf, 512, '\n');
                if ((*buf == ';') || (*buf == '#') || (*buf == '/'))
                        continue;
                if (*buf)
                {
                        int result = this -> Add (buf, portnum);
                        if (result == mwOk)
                                ++ counter;
                }
                if (!f)
                        return counter;
        }
} /* mwTask::Load */
 
 
// --------------------------------------------------
// -------------------- messages --------------------
// --------------------------------------------------
 
inline int mwTask::SendMessage (int fd, unsigned int ctrl,
        unsigned int code, const mwData &x)
{
        mwData sending;
        sending << ctrl << code << x. Length () << x;
        return mwSendBytes (fd, sending. Buffer (), sending. Length ());
} /* SendMessage */
 
inline int mwTask::RecvMessage (int fd, unsigned int &ctrl,
        unsigned int &code, mwData &x)
{
        // read the code and length of the message
        char buf00 [12];
        int result = mwRecvBytes (fd, buf00, 12);
        if (result < 0)
                return result;
        unsigned char *buf0 = reinterpret_cast<unsigned char *> (buf00);
 
        // extract the length of the message
        int len = (int) (buf0 [8]) << 24;
        len |= (int) (buf0 [9]) << 16;
        len |= (int) (buf0 [10]) << 8;
        len |= (int) (buf0 [11]);
        if (len < 0)
                return mwError;
 
        // extract the control code of the message
        ctrl = (int) (buf0 [0]) << 24;
        ctrl |= (int) (buf0 [1]) << 16;
        ctrl |= (int) (buf0 [2]) << 8;
        ctrl |= (int) (buf0 [3]);
 
        // extract the code of the message
        code = (int) (buf0 [4]) << 24;
        code |= (int) (buf0 [5]) << 16;
        code |= (int) (buf0 [6]) << 8;
        code |= (int) (buf0 [7]);
 
        // if the message length is zero, no more reading is necessary
        if (!len)
        {
                x. Reset ();
                return mwOk;
        }
 
        // prepare a memory buffer for the message
        char *buf = new char [len];
        if (!buf)
                return mwError;
 
        // read the message
        result = mwRecvBytes (fd, buf, len);
        if (result < 0)
        {
                delete [] buf;
                return result;
        }
 
        // transform the message to mw data
        x. Take (buf, len);
        return mwOk;
} /* RecvMessage */
 
inline int mwTask::QuitWorkers ()
{
        // write to the log file what you are doing
        if (logFile)
                *logFile << "Turning workers off..." << std::endl;
        
        // try connecting to each worker and ask them to exit
        int counter = 0;
        for (unsigned int n = 0; n < computers. size (); ++ n)
        {
                // retrieve the computer name and port from the list
                const char *name = computers [n]. c_str ();
                int port = ports [n];
 
                // if no valid name or port number read, skip this item
                if (!*name || !port)
                        continue;
 
                // try connecting to the computer
                int fd = mwConnect (name, port);
 
                // if unsuccessful, make a note and take another one
                if (fd < 0)
                {
                        if (logFile)
                        {
                                *logFile << "Worker " << name << ":" <<
                                        port << " could not be contacted." <<
                                        std::endl;
                        }
                        continue;
                }
                
                // prepare the control code to send to the worker
                unsigned int code = mwByeMsg | mwDontKeepMsg;
 
                // send the 'Bye!' message to the worker and disconnect it
                mwData empty;
                int result = this -> SendMessage (fd, ctrlnum, code, empty);
                mwDisconnect (fd);
 
                // add an appropriate message to the log file
                if (result == mwOk)
                {
                        ++ counter;
                        if (logFile)
                        {
                                *logFile << "Worker " << name << ":" <<
                                        port << " exited successfully." <<
                                        std::endl;
                        }
                }
                else if (logFile)
                {
                        *logFile << "Error while sending the disconnect "
                                "message to " << name << ":" << port <<
                                "." << std::endl;
                }
        }
 
        // write to the log file how many workers were turned off
        if (logFile)
                *logFile << counter << " worker(s) have been shut down." <<
                        std::endl;
 
        return mwOk;
} /* mwTask::QuitWorkers */
 
 
} // namespace multiwork
} // namespace chomp
 
#endif // _CHOMP_MULTIWORK_MWTASK_H_