DisplayServer.cc

Go to the documentation of this file.

//-------------------------------------------------------------------------
// File: DisplayServer.hh
//
// The DisplayServer class comprises methods to receive monitoring data from
// the consumers, store them and send them to the HistoDisplay clients.
// The communication between the consumers, the DisplayServer and the
// HistoDisplay is implemented with TSockets.
//
// Author: Wolfgang Wagner 
//
//-------------------------------------------------------------------------
//****************************************************************************
// RCS Current Revision Record
//-----------------------------------------------------------------------------
// $Source: /cvs/minoscvs/rep1/minossoft/CDFMonitoringFwk/DisplayServer.cc,v $
// $Revision: 1.10 $
// $Date: 2003/10/17 20:35:58 $
// $Author: dap56 $
// $State: Exp $
// $Locker:  $
//*****************************************************************************

#include "DisplayServer.h"

#include <unistd.h>
#include <iostream>
#include <cstring>

#include "TMessage.h"
#include "TString.h"

#include "MessageStorage.h"
#include "SocketUtils.h"
#include "ServerProtocol.h"

using std::string;
using std::cout;
using std::endl;
using std::cerr;


DisplayServer::DisplayServer(char* host, int consPort, TROOT* rt, 
  ConsumerList* cl, bool ut) : TNamed("DisplayServer","DisplayServer"),
  _hostName(host), 
  _consumerPort(consPort), _clientPort(9091), _error(DS_OK), _debugFlags(0x0),
  _updateTimeDiff(120000), _useTimer(ut), 
  _ss(0), _conSock(0), 
  _currInfo(0), _conList(cl), _glRoot(rt), _updateTimer(0)
{
  // Connect to the consumer server socket.
  _conSock = new TSocket((TString)(_hostName.c_str()), _consumerPort);

  if (_conSock->IsValid()) {
    cout << _clName << ": Connected successfully to CONSUMER socket at port "
         << _consumerPort << endl;
    cout << spaces << "on host " << _hostName << "." 
         << endl;
  }
  else {
    cerr << _clName << ": Consumer socket at port " << _consumerPort
         << " on host\n" << spaces << _hostName << " is not valid!" << endl;  
    _error = DS_NO_CONNECTION;
  } 
  
  if (_error < 0) ;
  else {
    _consMoni.Add(_conSock);
    if (openClientServerSocket() < 0) {
      cerr << _clName << ": ERROR could not open server socket for clients."
           << endl;
    }
  }
  // The following 3 lines are included to prevent linking problems. -------
  TConsumerInfo* currInfo = new TConsumerInfo("Display Server", 1);
  delete currInfo;
  currInfo = 0;
  //------------------------------------------------------------------------
}


const string DisplayServer::_clName = string("DisplayServer");


DisplayServer::~DisplayServer()
{
}


int DisplayServer::pollConsumer()
{
  static int consSendCycle=0;
  static int numConsPoll=0;   // Count the number of polls 
  static string oldNameString;
  TConsumerInfo* newInfo = 0;

  if (!_consMoni.GetActive()) {
    return 0;
  }

  TSocket* sock0 = _consMoni.Select(50);
  numConsPoll++;
  if (sock0 != (TSocket*)-1) {       
    if (sock0 && _conSock) {
      if (sock0->GetInetAddress().GetHostName() !=
          _conSock->GetInetAddress().GetHostName()) {
        cerr << "DisplayServer::pollConsumer(): Expected consumer "
             << "socket to be selected."
             << endl;
        return -4;
      }
    }      
    // For debugging:
    // cout << "Returned with valid sock0. # of polls = " 
    //      << numConsPoll << endl;
    // Reset the number of polls if the connection was successful.
    numConsPoll=0;
    // Process message
    SocketUtils sockUt0(_glRoot, &_consMoni, sock0);
    string ctrlString = sockUt0.getStringMessage();
    // For debugging only:---------------------------
    // cout << _clName << ": Got string: " << ctrlString << endl;
    //-----------------------------------------------
    if (ctrlString == DspEndConnection) {
      string consumerName; 
      if (_currInfo) consumerName = string(_currInfo->GetTitle());
      else           consumerName = "Consumer";
      cout << _clName << ": INFO: " << consumerName << " sent message: "
           << ctrlString << "\n" << spaces              
           << "==> Going to terminate connection to " << consumerName;
      cout << " running on\n" 
           << spaces << sock0->GetInetAddress().GetHostName() << endl;
      sockUt0.closeSocket();
      //---------------------------------------------
      _conList->setStatus(0,ConsumerList::Finished);
      //      _conList->sendList("b0dap30.fnal.gov",9090);
      sleep(1);
      if (_ss) {
        _ss->Close();
        _glRoot->GetListOfSockets()->Remove(_ss);
        delete _ss;
        _ss = 0;
      }
      //---------------------------------------------       
      // Now there should be only the client sockets left.
      TIter next(_glRoot->GetListOfSockets());
      TSocket* socktmp = 0;
      while ((socktmp = (TSocket*)next())) {
        socktmp->Send(DspEndConnection.c_str());
        sleep(1);
        socktmp->Close();
        _glRoot->GetListOfSockets()->Remove(socktmp);
      }
      return 2;
    }
    else {
      if (ctrlString == DspConsumerSend) {
        consSendCycle++;
        // For debugging:
        // cout << "Send Cycle: " << consSendCycle << endl; 
        _glRoot->DeleteAll();
        TMessage* conmess = 0;
        //---------------------------------------------------------
        // 2.) Receive the TConsumerInfo 
        sock0->Recv(conmess);  
        if (conmess) { 
          if (conmess->What() == kMESS_OBJECT) {
            TObject *rcvobject = (TObject*)conmess->ReadObject( 
                                 conmess->GetClass());
            // For debugging: -------------------------------
            // cout << _clName << ": objectname = " 
            //     << rcvobject->GetName() << endl;
            //-----------------------------------------------
            if(rcvobject->InheritsFrom("TConsumerInfo")) {
              newInfo = (TConsumerInfo*) rcvobject;
              if (newInfo) {
                delete _currInfo;
                _currInfo = newInfo;
              }  
            }
            else {
              cerr << _clName << ": pollConsumer(): ERROR: "
                   << "Did not receive the TConsumerInfo!"
                   << endl;
              return -1;
            }           
          }
          delete conmess;
        }  
        else {
          // Error:
          cerr << _clName << ": pollConsumer(): ERROR: "
               << "Didn't receive a valid message." << endl;        
          return -2; 
        }  
        //---------------------------------------------------------
        // 3.) Receive the string with the object names.
        string namesString = sockUt0.getStringMessage();
        // For Debugging: 
        // cout << "Got names-String: " << namesString << endl;
        if (strncmp(namesString.c_str(), "Modified", 8) == 0) {
          if (_debugFlags & 0x4) {
            cout << _clName << ": pollConsumer(): DEBUG INFO: "
                 << "Going to edit the storage list." << endl;
          }             
          editStorageList(namesString);
          setRequireUpdateBits(); 
        }
        else { 
          if (strncmp(namesString.c_str(), "Unmodified", 10) != 0) {
            cerr << _clName << ": pollConsumer(): ERROR: "
                 << "Expected key word 'Unmodified'." << endl;
            return -3;
          }
        }  
        //-----------------------------------------------------
        // 4.) Receive the objects expect to fill the list.
        if (_currList.GetSize() > 0) {
          TListIter storeIter(&_currList);
          Int_t     recvRes = 0;
          while (MessageStorage *mesStor = (MessageStorage*)storeIter()) {
            if (sock0) recvRes = sock0->Recv(conmess);
            if ((recvRes != -1) && (recvRes != 0)) { 
              if (conmess->What() == kMESS_OBJECT) {
                mesStor->updateMessage(conmess);
                // For Debugging:
                // cout << _clName << ": pollConsumer(): INFO: "
                //      << "stored " << mesStor->GetName() << endl;
              }
              else {
                cerr << _clName << ": pollConsumer(): Expected an object."
                     << endl;
                continue;
              }
            }
            else {
              cerr << _clName << ": ERROR reading object messages!" << endl;
              cerr << _clName << ": Closing socket!" <<  endl;
              sockUt0.closeSocket();
              continue;
            }
            // Make sure that conmess does not point anymore to
            // the objects in the storage list.
            conmess = 0;   
          }
        }  
        else {
          cerr << _clName << ": pollConsumer(): Storage list empty."
               << endl;
        }
        //---------------------------------------------------------
        // 5.) Wait for the end string to be sent.
        string endString = sockUt0.getStringMessage();
        if (endString != DspConsumerFinish)
          cerr << _clName << ": pollConsumer(): ERROR: "
               << "Expected different end string. Got " << endString
               << endl;
        //----------------------------------------------------
        if (consSendCycle == 1) {
          updateStateManager();
          if (!_updateTimer && _useTimer) {
            _updateTimer = new TTimer(this, _updateTimeDiff);
            _updateTimer->Reset();
            _updateTimer->TurnOn(); 
          }
        }
      }
      else {
        //        cerr << _clName << ": Got an unexpected string from the "
        //   << "consumer." << endl;
      }
    } // else of if (string=="END CONNECTION")
  } 

  return 1;
}


void DisplayServer::connectClient()
{
  // Check for requests by the display clients 
  TSocket* sock1=0;
  sock1 =  _ss->Accept();
  if (sock1 != (TSocket*)-1) {
    string clientHostname(sock1->GetInetAddress().GetHostName());
    int  clientPort = sock1->GetInetAddress().GetPort();
    cout << _clName << ": connectClient(): Client connected from " 
         << clientHostname << " on port " << clientPort << "." << endl;
    _clientMoni.Add(sock1);
    char portChar[20];
    sprintf(portChar, "%d", clientPort);
    string id = clientHostname + string(":") + string(portChar);
    _clientMap[id] = false;
  }

  return;
}


void DisplayServer::pollClients()
{
  const string myName("DisplayServer::pollClients()");

  if (!_clientMoni.GetActive()) return;

  // wait for message 
  TSocket* sock2 = 0;
  sock2 = _clientMoni.Select(50);
  if (sock2 != (TSocket*)-1) {
    SocketUtils sockUt2(_glRoot, &_clientMoni, sock2);
    string request = sockUt2.getStringMessage();
    // For debugging purposes:
    // cout << _clName << ": Got requestString: " << request << endl;
    // cout << "                Size = " << request.size() << endl;
    if (request.size() == 0) {
      cerr << myName << ": request empty" << endl;
      return;
    }  
    if (request == DspEndConnection) {
      cout << myName << ": Terminating connection to "  
           << sockUt2.getHostname() << "." << endl;
      sockUt2.closeSocket();
#ifndef IRIX6
      _clientMap.erase(sockUt2.getId());
#else
      // erase entry from map
      // SGI barfs at _clientMap.erase(sockUt2.getId());
      map<string,bool>::iterator cm_itr = _clientMap.find(sockUt2.getId());
      _clientMap.erase(cm_itr);
#endif
      return;
    }  
    if (request == "ConsumerInfo") {  
      TMessage infMess(kMESS_OBJECT);
      infMess.Reset();
      infMess.WriteObject(_currInfo);
      int sendRes = 0;
      sendRes = sock2->Send(infMess);
      if (sendRes == -1) {
        cerr << myName << "Error sending TConsumerInfo." << endl;
        sockUt2.closeSocket();
      } 
      if (_clientMap[sockUt2.getId()]) _clientMap[sockUt2.getId()] = false;      
    }
    else {
      // Assume the received requestString is a request for an object.
      // Get object by name and send it
      if (_clientMap[sockUt2.getId()]) {
        Int_t sendRes=0;
        sendRes = sock2->Send(DspRequestNewInfo.c_str());
        if ( sendRes == -1) {
          cerr << myName << ": ERROR: Socket->Send returned "
               << "with " << sendRes << ".\n" 
               << "               ==> The socket will be closed."
               << endl;
          // Close the socket in case of an error.
          sockUt2.closeSocket();
        }
      }        
      else {
        Bool_t objFound = kFALSE; 
        TListIter iter(&_currList);
        while (MessageStorage* obj = (MessageStorage*) iter()) {
          if (obj->GetName() == request) {
            objFound = kTRUE;  
            TMessage sendMess(kMESS_OBJECT);
            TMessage* storeMess = obj->getMessage();
            Int_t size = storeMess->BufferSize();
            Int_t bufSize = size+200;  
            char* buffer = new char[bufSize];
            // For debugging: 
            // cout << "Buffer Size:" << storeMess->BufferSize() << endl;
            storeMess->ReadBuf(buffer,bufSize);
            storeMess->Reset();
            sendMess.WriteBuf(buffer,bufSize);
            Int_t sendRes=0;
            sendRes=sock2->Send(sendMess);
            if ( sendRes == -1) {
              cerr << myName << ": ERROR: pollClients(): "
                   << "Socket->Send returned with " << sendRes << ".\n" 
                   << "               ==> The socket will be closed."
                   << endl;
              // Close the socket in case of an error.
              sockUt2.closeSocket();
            }
            // For debugging:                
            // cout << "SendRes = " << sendRes << endl;
            delete [] buffer;
            continue;
          }
        }       
        if (!objFound) {
          Int_t sendRes=0;
          sendRes = sock2->Send(DspObjectNotFound.c_str());
          if ( sendRes == -1) {
            cerr << myName << ": ERROR: Socket->Send returned "
                 << "with " << sendRes << ".\n" 
                 << "               ==> The socket will be closed."
                 << endl;
            // Close the socket in case of an error.
            sockUt2.closeSocket();
          }
        }
      }          
    } // else of if (request == "ConsumerInfo")
  } // if (sock2 != (TSocket*)-1)
  
  return;
}


TConsumerInfo* DisplayServer::getCurrentInfo()
{
  return _currInfo;  
}


void DisplayServer::editStorageList(const string& nameString)
{
  const string myName("DisplayServer::editStorageList()");
  char* help1 = 0;
  char* help2 = 0;

  // Add those objects to "oldList" which are in list but not named
  // in nameString anymore.
  if (_debugFlags & 0x8) {
    cout << "Outdated Size = " << _outdatedList.GetSize() << "  list size = "
         << _currList.GetSize() << endl;
  }  
  TListIter iter1(&_currList);
  while (MessageStorage* obj = (MessageStorage*) iter1()) {
    Bool_t found = kFALSE;
    // For debugging:
    //cout << myName << obj->GetName() << endl;
    //Find the first $:
    char* cpyName = new char[nameString.size()+1];
    strcpy(cpyName, nameString.c_str());
    help1 = const_cast<char *>(strchr(cpyName,'$'));
    while ((help2 = const_cast<char*>(strtok(help1, "$"))) && (!found)) {
      if (obj->GetName() == string(help2)) found = kTRUE;        
      help1 = 0; 
    }
    if (!found) {
      // Check whether an object of this type exists already in
      // oldList.
      TListIter iterOld(&_outdatedList);
      Bool_t oldFound = kFALSE;
      while (MessageStorage* oldObj = (MessageStorage*) iterOld()) {
        if (oldObj->GetName() == obj->GetName()) {
          oldObj->updateMessage(obj->getMessage());
          oldFound = kTRUE;
          continue;
        }      
      }
      if (!oldFound) {
        cout << _clName << "::editStorageList(): INFO: The object '"
             << obj->GetName() << "' was removed\nfrom the object list"
             << " by the consumer.\nThe last copy is kept by the Display"
             << " Server." << endl; 
        _outdatedList.AddLast(obj);
      } 
    }
  }       
  _currList.Delete();
  if (_currList.GetSize() > 0) {
    cerr << myName << ": list size should be 0." << endl;
  }  
  // Extract the names of the Canvases, which are supposed to be separated
  // by $ signs.
  // Build the list new.
  //Find the first $:
  string cpy2Name(nameString);
  help1 = const_cast<char*>(strchr(cpy2Name.c_str(),'$'));
  while ((help2 = const_cast<char*>(strtok(help1, "$")))) {
    if (_debugFlags & 0x2) {
      cout << myName << ": Found String: " << help2 << endl;
    }  
    MessageStorage *storage = new MessageStorage(help2);
    _currList.AddLast(storage);    
    help1 = 0; 
  } 

  return;
}


void DisplayServer::updateStateManager()
{
  // Fill consumer list and send the list to the state manager. 

  const string myName("DisplayServer::updateStateManager()");

  if (_currInfo) {
    _conList->addEntry(_currInfo->GetTitle(), (TString)(_hostName.c_str()),
                       _clientPort, _currInfo->nevents(), _currInfo->runnumber(),
                       ConsumerList::Running);
    //conslist.print();
    if (_debugFlags & 0x10)
      cout << myName << ": Consumer information will be sent to\n"
           << spaces << "the State Manager running on b0dap30.fnal.gov" << endl;
    // Send list to StateManager
    //    _conList->sendList("b0dap30.fnal.gov",9090);
  }
  else {
    cerr << myName << ": Consumer Info not available." << endl;
    _conList->addEntry("Unknown", (TString)(_hostName.c_str()), _clientPort,
                       0, 0, ConsumerList::Running);
    //    _conList->sendList("b0dap30.fnal.gov",9090);
  }

  return;
}


Bool_t DisplayServer::HandleTimer(TTimer* /* timer */)
{
  updateStateManager();

  return kTRUE;
}


void DisplayServer::setRequireUpdateBits()
{
  map<string,bool>::iterator it;
  for (it=_clientMap.begin(); it!=_clientMap.end(); ++it) {
    //string name = (*it).first;
    //_clientMap[name] = true;
    (*it).second = true;
  }
  return;
}


int DisplayServer::openClientServerSocket() 
{
  const string myName("DisplayServer::openClientServerSocket()");

  int rtVal = DS_OK;

  // Open a server socket for the display clients to connect 
  _ss = new TServerSocket(_clientPort, kTRUE);
  // Open a server socket for the clients, testing different ports
  while (! _ss->IsValid()) {  
    delete _ss;
    _clientPort++;
    _ss = new TServerSocket(_clientPort, kTRUE);
  }
  if (!(_ss->IsValid())) {
    cout << myName << ": server socket not valid." << endl;
    _error = DS_BAD_CLIENT_SSOCKET;
    rtVal  = DS_BAD_CLIENT_SSOCKET;
  }
  else {    
    cout << barrier << "\n\n" 
         << _clName << ": Server socket for HISTO DISPLAY clients at PORT " 
         << _clientPort << endl; 
    cout << spaces << "available.\n\n"
         << spaces << "Use   " << _hostName << ":" << _clientPort << endl; 
    cout << spaces << "as input stream to connect from the HistoDisplay" << endl;
    cout << spaces << "to the Display Server.\n\n"  
         << barrier << "\n" 
         << endl;
    // Disable blocking mode for this server socket  
    _ss->SetOption(kNoBlock,1);  
  }

  return rtVal;
} 

---