• Main
• Home
• Features
• Screenshots
• Downloads
• License
• Team profile
• Documentation
• Manual
• FAQ
• Reference

Search this page:

PCMClient.cpp

Go to the documentation of this file.

//
// This source file is a part of ParCompMark
// Parallel Compositing Benchmark Framework
//
// for latest info see http://parcompmark.sourceforge.net

//
// Copyright (C) 2006 IT2 ParCompMark Dev. Team
// 
// This program 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 2
// of the License, or (at your option) 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 General Public License for more details.
// 
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.

//
// Inner includes
//

#include "../include/PCMClient.h"

#include "../include/PCMApplication.h"

namespace ParCompMark
{

  //
  // Constructors & destructor
  //

  Client::Client(const std::string & name):
        // Parent initializer 
  Network(name          /* Name of the class */
        )
        // You have to initialize the following attributes:
        // - mServerIP
        // - mType
        // - mParent
        // - mMessage
  {
        mParent = 0;
  }

 /*----------------------------------------------------------------------*/

  Client::~Client()
  {
  }

 /*----------------------------------------------------------------------*/

  //
  // Methods
  //

  void Client::openConnection()
  {
        Assert(mStreamSocket == -1, INVALID_OPERATION_ERROR, "Client::openConnection()");
        Assert(mServerIP != "", INVALID_OPERATION_ERROR, "Client::openConnection()");

        struct sockaddr_in addr;

        if(mStreamSocket != -1)
        {
         return;
        }

        if((mStreamSocket = socket(AF_INET, SOCK_STREAM, 0)) < 0)
        {
         Except(INTERNAL_ERROR, "Client::openConnection()", "Unable to create socket");
        }

        addr.sin_family = AF_INET;
        addr.sin_port = htons(mCommunicationPort);
        addr.sin_addr.s_addr = inet_addr(mServerIP.c_str());
        memset(&(addr.sin_zero), '\0', 8);

        if((connect(mStreamSocket, (struct sockaddr *) &addr, sizeof(struct sockaddr))) < 0)
        {
         Except(INTERNAL_ERROR, "Client::openConnection()", "Unable to connect socket");
        }

        Logger::getInstance()->log(Logger::NOTICE, "TCP/IP connaction is opened.");
  }

 /*----------------------------------------------------------------------*/

  void Client::closeConnection()
  {
        Assert(mStreamSocket != -1, INVALID_OPERATION_ERROR, "Client::closeConnection()");

        close(mStreamSocket);

        mStreamSocket = -1;
  }

 /*----------------------------------------------------------------------*/

  void Client::sendMessage(const std::string & type, const std::string & message)
  {
        Assert(mStreamSocket != -1, INVALID_OPERATION_ERROR, "Client::sendMessage()");

        u32 typeLength,
        messageLength;

        std::string m = message;

        typeLength = (u32) type.length();
        messageLength = (u32) m.length();

        if(messageLength == 0)
        {
         m = "NONE";
         messageLength = m.length();
        }

        u8 buff[4];

        buff[0] = (u8) ((typeLength & 0xFF000000) >> 24);
        buff[1] = (u8) ((typeLength & 0x00FF0000) >> 16);
        buff[2] = (u8) ((typeLength & 0x0000FF00) >> 8);
        buff[3] = (u8) ((typeLength & 0x000000FF) >> 0);

        if(send(mStreamSocket, buff, 4, 0) != 4)
        {
         Except(INTERNAL_ERROR, "Client::sendMessage()", "Unable to send message");
        }

        if(send(mStreamSocket, type.c_str(), typeLength, 0) != typeLength)
        {
         Except(INTERNAL_ERROR, "Client::sendMessage()", "Unable to send message");
        }

        buff[0] = (u8) ((messageLength & 0xFF000000) >> 24);
        buff[1] = (u8) ((messageLength & 0x00FF0000) >> 16);
        buff[2] = (u8) ((messageLength & 0x0000FF00) >> 8);
        buff[3] = (u8) ((messageLength & 0x000000FF) >> 0);

        if(send(mStreamSocket, buff, 4, 0) != 4)
        {
         Except(INTERNAL_ERROR, "Client::sendMessage()", "Unable to send message");
        }

        if(send(mStreamSocket, m.c_str(), messageLength, 0) != messageLength)
        {
         Except(INTERNAL_ERROR, "Client::sendMessage()", "Unable to send message");
        }

        Logger::getInstance()->log(Logger::NOTICE, "TCP message sent: " + type + "|" + message);

  }

 /*----------------------------------------------------------------------*/

  void Client::recieveMessage()
  {
        Assert(mStreamSocket != -1, INVALID_OPERATION_ERROR, "Client::recieveMessage()");

        std::string type, message, temp, str;

        bool end = false;

        int recvLength,
        index;

        u32 length = 0;

        recvLength = -1;

        u32 sumRecieved;

        char *buff;
        u8 *lengthBuff = new u8[4];

        //type length and type

        sumRecieved = 0;
        while(sumRecieved != 4)
        {

         if((recvLength = recv(mStreamSocket, lengthBuff + (sumRecieved), 4 - sumRecieved, 0)) < 0)
         {
                Except(INTERNAL_ERROR, "Client::recieveMessage()", "Unable to recieve message");
         }
         sumRecieved += recvLength;
        }

        length = length | (lengthBuff[0] << 24);
        length = length | (lengthBuff[1] << 16);
        length = length | (lengthBuff[2] << 8);
        length = length | (lengthBuff[3] << 0);

        buff = new char[length + 1];

        sumRecieved = 0;
        while(sumRecieved != length)
        {

         if((recvLength = recv(mStreamSocket, buff + (sumRecieved), length - sumRecieved, 0)) < 0)
         {
                Except(INTERNAL_ERROR, "Client::recieveMessage()", "Unable to recieve message");
         }
         sumRecieved += recvLength;
        }

        if(length != sumRecieved)
        {
         Except(INTERNAL_ERROR, "Client::recieveMessage()",
                 "Bad message length. Sended: " + StringConverter::toString(length) + ", Recieved: " +
                 StringConverter::toString(sumRecieved));
        }

        buff[length] = '\0';

        type = buff;

        //message length and message

        sumRecieved = 0;
        while(sumRecieved != 4)
        {

         if((recvLength = recv(mStreamSocket, lengthBuff + (sumRecieved), 4 - sumRecieved, 0)) < 0)
         {
                Except(INTERNAL_ERROR, "Client::recieveMessage()", "Unable to recieve message");
         }
         sumRecieved += recvLength;
        }

        length = 0;

        length = length | (lengthBuff[0] << 24);
        length = length | (lengthBuff[1] << 16);
        length = length | (lengthBuff[2] << 8);
        length = length | (lengthBuff[3] << 0);

        delete buff;
        buff = new char[length + 1];

        sumRecieved = 0;
        while(sumRecieved != length)
        {

         if((recvLength = recv(mStreamSocket, buff + (sumRecieved), length - sumRecieved, 0)) < 0)
         {
                Except(INTERNAL_ERROR, "Client::recieveMessage()", "Unable to recieve message");
         }
         sumRecieved += recvLength;
        }

        if(length != sumRecieved)
        {
         Except(INTERNAL_ERROR, "Client::recieveMessage()",
                 "Bad message length. Sended: " + StringConverter::toString(length) + ", Recieved: " +
                 StringConverter::toString(sumRecieved));
        }

        buff[length] = '\0';

        message = buff;

        delete buff;

        mMessage = message;
        mType = type;

        Logger::getInstance()->log(Logger::DEBUG, "message recieved: " + mType + "|" + mMessage);

  }

 /*----------------------------------------------------------------------*/

  void Client::handleMessage()
  {

        //printf("\n%s - %s\n", mType.c_str(), mMessage.c_str());

        Logger::getInstance()->log(Logger::DEBUG, "message handle start: " + mType + "|" + mMessage);

        if((Host::getInstance()->getProcessCount() == 0)
         && (mType != "LOGIN" && mType != "LOGINOK" && mType != "INIT" && mType != "INITOK" && mType != "QUIT"
           && mType != "QUITOK" && mType != "LOWLEVELSCRIPT" && mType != "LOWLEVELSCRIPTOK" && mType != "TIME"
           && mType != "TIMEOK"))
        {

         //fprintf(stderr, "NOACTIVE\n");

         openConnection();
         sendMessage("NOACTIVE", "NONE");
         closeConnection();

        } else
        {
         //LOGIN (source: server(broadcast); answer: LOGINOK; handle: NetClient)
         if(mType == "LOGIN")
         {
                openConnection();
                Network::getIP();
                sendMessage("LOGINOK", Network::getHostName());
                recieveMessage();
                handleMessage();
                closeConnection();
                //LOGINOK (source: client(TCP/IP); answer: ; handle: HandleClient)
         } else if(mType == "LOGINOK")
         {
                struct sockaddr_in addr;
                unsigned int addrLen = sizeof(struct sockaddr);

                if(getpeername(mStreamSocket, (struct sockaddr *) &addr, &addrLen) < 0)
                {
                 Except(INTERNAL_ERROR, "Client::handleMessage()", "Unable to get peer name.");
                }

                std::string ip = inet_ntoa(addr.sin_addr);

                unsigned int id = 100000;

                Cluster::getInstance()->getHosts().lock();
                if(!Cluster::getInstance()->getHosts()->has(mMessage))
                {
                 HostInfo::Pointer h(new HostInfo(ip, OutputNode::INFORMATION));

                 Cluster::getInstance()->getHosts()->add(mMessage, h);

                 id = Cluster::getInstance()->getHosts()->getSize();
                }
                Cluster::getInstance()->getHosts().unlock();

                sendMessage("ID", StringConverter::toString(id));

         } else if(mType == "ID")
         {

                unsigned int id = StringConverter::toU32(mMessage);

                if(id < 100000)
                 Host::getInstance()->setID(id);

         }
         //QUIT (source: server(broadcast); answer: QUITOK; handle: NetClient)
         else if(mType == "QUIT")
         {
                openConnection();
                sendMessage("QUITOK", "NONE");
                closeConnection();
                stopThread();
         }
         //QUITOK (source: client(TCP/IP); answer: ; handle: HandleClient)
         else if(mType == "QUITOK")
         {

         }
         //INIT (source: server(broadcast); answer: INITOK; handle: NetClient)
         else if(mType == "INIT")
         {

                openConnection();
                sendMessage("INITOK", Application::getInstance()->getHostInfo()->serialize2XML());
                recieveMessage();       //LOWLEVELSCRIPT
                handleMessage();
                closeConnection();
         }
         //INITOK (source: client(TCP/IP); answer: ; handle: HandleClient)
         else if(mType == "INITOK")
         {
                //colect cluster information
                OutputNode::Pointer hostinfo(Application::getInstance()->getCollectClusterDescription());
                hostinfo.lock();
                hostinfo->createChildNode(mMessage.c_str());
                hostinfo.unlock();

                sendMessage("LOWLEVELSCRIPT", Application::getInstance()->getLowLevelScript());
                recieveMessage();       //LOWLEVELSCRIPTOK
                handleMessage();
         }
         //STOP (source: handleclient(TCP/IP); answer: LOWLEVELSCRIPTOK; handle: NetClient)
         else if(mType == "LOWLEVELSCRIPT")
         {

                Network::getIP();

                Host::getInstance()->setName(Network::getHostName());
                Host::getInstance()->setLowLevelScript(mMessage);
                Host::getInstance()->initialize();
                Host::getInstance()->start();

                sendMessage("LOWLEVELSCRIPTOK", "NONE");
         }
         //LOWLEVELSCRIPTOK (source: client(TCP/IP); answer: ; handle: HandleClient)
         else if(mType == "LOWLEVELSCRIPTOK")
         {
                ;
         }
         //STOP (source: server(broadcast); answer: STOPOK; handle: NetClient)
         else if(mType == "STOP")
         {
                u32 frameID = Host::getInstance()->stop();

                openConnection();
                sendMessage("STOPOK", StringConverter::toString(frameID));
                closeConnection();

         }
         //STOPOK (source: client(TCP/IP); answer: ; handle: HandleClient)
         else if(mType == "STOPOK")
         {
                u32 frameID = StringConverter::toU32(mMessage);

                NetServer::IntPointer stopID = ((NetServer *) mParent)->getStopFrameID();

                stopID.lock();
                if(*stopID.getPtr() < frameID)
                {
                 *stopID.getPtr() = frameID;
                }
                stopID.unlock();

         }
         //FRAMEID (source: server(broadcast); answer: FRAMEIDOK; handle: NetClient)
         else if(mType == "FRAMEID")
         {
                u32 stopID = StringConverter::toU32(mMessage);

                Host::getInstance()->setFrameID(stopID);

                Host::getInstance()->collectData();
                std::string statistic = Host::getInstance()->getOutputDocument()->serialize2XML();
                openConnection();
                sendMessage("FRAMEIDOK", statistic);
                closeConnection();
         }
         //FRAMEIDPOK (source: client(TCP/IP); answer: ; handle: HandleClient)
         else if(mType == "FRAMEIDOK")
         {
                // Append statistics to the gathered data
                OutputNode::Pointer outputNode(Application::getInstance()->getCurrentExecutionOutputDocument());
                outputNode.lock();
                outputNode->createChildNode(mMessage.c_str());
                outputNode.unlock();

         }
         //COLLECTDATA (source: server(broadcast); answer: COLLECTDATAOK; handle: NetClient)
         else if(mType == "COLLECTDATA")
         {

                Host::getInstance()->collectData();
                openConnection();
                sendMessage("COLLECTDATAOK", Host::getInstance()->getOutputDocument()->serialize2XML());
                closeConnection();

         } else if(mType == "COLLECTDATAOK")
         {
                // Append statistics to the gathered data
                OutputNode::Pointer outputNode(Application::getInstance()->getCurrentExecutionOutputDocument());
                outputNode.lock();
                outputNode->createChildNode(mMessage.c_str());
                outputNode.unlock();
         } else if(mType == "TIME")
         {

          Timer:sleep(Host::getInstance()->getID());
                openConnection();
                Real t1 = Timer::getUncorrectedSystemTime();

                sendMessage("TIMEOK", "NONE");
                recieveMessage();

                Host::getInstance()->setMessageSendingTime((Timer::getUncorrectedSystemTime() - t1) / 2);

                closeConnection();
         } else if(mType == "TIMEOK")
         {
                sendMessage("TT", "NONE");
         }
        }
  }

 /*----------------------------------------------------------------------*/

}

© Copyright - 2006 (IT)² ParCompMark Dev. Team