DDSParentServer.cxx

Go to the documentation of this file.

//                                                                           //
// DDSParentServer                                                           //
//                                                                           //
// Package: DDS (Data Dispatcher System).                                    //
//                                                                           //
// S. Kasahara 2/2001                                                        //
//                                                                           //
// Purpose: This class is part of the MINOS data dispatcher system.  It is   //
//          the main class of the parent server, which resides on the same   //
//          local network as the online data generator.  The DDSParentServer //
//          handles the setup of a ROOT TServerSocket to listen for          //
//          connections from potential clients.  When a client connection is //
//          accepted, DDSParentServer spawns a child server (managed by      //
//          DDSChildServer class) to handle the client data requests.        //
//                                                                           //
// Based in part on ideas from the LHCb OTT online monitoring system.        //

#include <cstdlib>
#include <unistd.h>      // fork used to spawn child process
#ifndef MACOSX
  #include <wait.h>      // waitpid used to check status of exited children
#else
  #include <sys/wait.h>  // waitpid used to check status of exited children
#endif
#include <errno.h> // errno used to catalog errors in system calls

#include "TMessage.h"  
#include "TSystem.h"

#include "MessageService/MsgService.h"

#include "Dispatcher/DDS.h"
#include "Dispatcher/DDSParentServer.h"
#include "Dispatcher/DDSClientId.h"

std::ostream& operator << (std::ostream& ms, DDSParentServer* ps) 
                                    {return ps->Print(ms);}

ClassImp(DDSParentServer)

// Definition of static data members
// *********************************

CVSID("$Id: DDSParentServer.cxx,v 1.16 2009/02/28 21:46:12 gmieg Exp $");

// Definition of methods (alphabetical order)
// ******************************************

//_____________________________________________________________________________
bool DDSParentServer::Authorize() const {
  // Purpose: Authorize currently connected client to use dispatcher.
  //          Address of currently connected client must have been previously
  //          stored in fClientAddress.
  // 
  // Argument: none.
  //
  // Return: true if client is authorized, else false.
  //
  // Contact: S. Kasahara
  //
  // Notes: Currently a dummy routine, always returns true.
  //

  // perform check on client address, if ok then
  return true;
 
}

//_____________________________________________________________________________
UInt_t DDSParentServer::CheckChildStatus() {
  // Purpose: Check status of any exited child processes spawned by
  //          parent.  This has the important side benefit of clearing 
  //          "zombies" (child processes that have exited but have status
  //          values left to be read).  The fNumChild counter of currently
  //          active children is decremented for each newly exited child.
  // 
  // Argument: none.
  //
  // Return: Number of remaining active child processes.  
  //
  // Contact: S. Kasahara
  //
  // Notes: Another way to do this would be for the parent server to maintain
  //        a list of spawned child processes and inquire each child's status
  //        according to its process id.

  int childpid;  // pid number of exited child processes
  Int_t childstat; // exit status of child process

  bool newexit = false;
  while ((childpid = waitpid(-1, &childstat, WNOHANG)) > 0) {
    // This child has exited
    if ( !newexit ) {
      newexit = true; 
      MSG("DDS",Msg::kInfo) 
      << "PS: ParentServer reporting on status of exited children: " << endl;
    }

    fNumChild--;  // decrement number of active children

    if ( childstat != 0 ) {
      // the child process exited with an error status
      MSG("DDS",Msg::kWarning) << "PS: Child process " << childpid << 
        " has exited with error status " << WEXITSTATUS(childstat) 
        << "." << endl;
    }
    else {
      // if in info mode report all child exits
      MSG("DDS",Msg::kInfo) << "PS: Child process " << childpid <<
        " has exited successfully." << endl;
    }
    fPSStatus.RemoveClientId(childpid);
  } 

  if ( newexit ) {
    MSG("DDS",Msg::kInfo) << "PS: Number of children currently running = " 
                          << fNumChild << "." << endl;
  }

  return fNumChild;

}


//_____________________________________________________________________________
DDSParentServer::DDSParentServer(UInt_t port, UInt_t maxchild, Int_t logLevel,
                                 Int_t maxInactive, Int_t backlog):
fTServerSocket(0),fMaxChild(maxchild),fLogLevel(logLevel),
fMaxInactive(maxInactive),fNumChild(0),fShutdown(false),fTSocket(0),
fMessageIn(0), fMessageOut(), fPSStatus() {
  // Purpose: Constructor.  This constructor creates a TServerSocket 
  //          attached to the specified port. A TServerSocket is created
  //          and bound to the specified port.
  // 
  // Arguments: port: port number assigned to server socket.  If port = 0
  //                  (default), the serversocket will be connected to the
  //                  named service "MinosDDS", which should be assigned to
  //                  a port in the /etc/services file.
  //            maxchild: maximum number of concurrent child servers allowed
  //                      (default = kMaxChild).
  //            logLevel: message verbose level
  //            maxInactive: max inactive time before connection shutdown
  //            backlog: maximum number of clients the kernel should queue
  //                     up on this port for pending connections (default is
  //                     TServerSocket::kDefaultBacklog).
  //
  // Return: n/a.
  //
  // Contact: S. Kasahara
  //
  // Notes: Use IsValid() to check if DDSParentServer was created successfully.
  //

  // Create server socket
  // The second argument kTRUE sets the SO_REUSEADDR socket option 
  // (kFALSE by default). When set true, reuse allows the listening server to 
  // start and bind its port even if previously established connections 
  // exist that use this port as their local port (such as if the parent
  // starts, spawns a child, dies, and needs to be restarted with the child
  // still serving from that port). 
  if (!port) {
    // User has not specified an alternative port.  server socket will be
    // bound to the port assigned to the named service "MinosDDS" 
    fTServerSocket = new TServerSocket("MinosDDS",kTRUE,backlog);
  }
  else {
    fTServerSocket = new TServerSocket(port,kTRUE,backlog);
  }

  // Check validity of socket
  if(!fTServerSocket || !fTServerSocket -> IsValid()) {
    // An error occured during the creation of the server socket

    if ( port ) {
      MSG("DDS",Msg::kWarning) 
        << "PS: Unable to create parent server socket connected to port " 
        << port << "." << endl;
    }
    else {
      MSG("DDS",Msg::kWarning) 
    << "PS: Unable to create parent server socket connected to named service " 
        << "MinosDDS." << endl;
    }
    if(fTServerSocket)delete fTServerSocket; fTServerSocket = 0;
    fShutdown = true;
  } 
  else {
    // Store local address of the server socket.
    fServerAddress = fTServerSocket -> GetLocalInetAddress();
  }

}

//_____________________________________________________________________________
DDSParentServer::~DDSParentServer() {
  // Purpose: Destructor.  
  // 
  // Arguments: n/a.
  //
  // Return: n/a.
  //
  // Contact: S. Kasahara
  //
  
  if ( IsValid() ) {
    delete fTServerSocket; fTServerSocket = 0;
  }
  // fTSocket should always be deleted and cleared in DDSParentServer::Run() 
  if ( fTSocket ) delete fTSocket; fTSocket = 0;
    
}

//_____________________________________________________________________________
bool DDSParentServer::IsLocalClient() const {
  // Purpose: Determine if currently connected client with address given
  //          in fClientAddress is local.
  // 
  // Arguments: none.
  //
  // Return: true if local.
  //
  // Contact: S. Kasahara
  //
  // Notes:  This method currently determines that a client is local only if it
  //         resides on the same host as the parent server. The definition
  //         of a local client will eventually be expanded to include other 
  //         hosts in the same local area network (once I know how to do that).
  //         Having local client status bestows special privileges on that
  //         client, such as:
  //         a) local client child servers are run at higher priority than 
  //            remote client child servers.
  //         b) only local clients may shutdown the parent server.
  //

  bool local = false;

  TString localhost(gSystem->GetHostByName(gSystem->HostName()).GetHostName());
  
  if(!localhost.CompareTo(fClientAddress.GetHostName(),TString::kIgnoreCase)){ 
    local = true;
  }

  return local;

}

//_____________________________________________________________________________
std::ostream& DDSParentServer::Print(std::ostream& ms) const {
  // Purpose: Print DDSParentServer status on std::ostream.
  // 
  // Arguments: ms  std::ostream to print on.
  //
  // Return: std::ostream reference.
  //
  // Contact: S. Kasahara
  //

  if ( IsValid() ) {
    ms << "PS: local addr " 
    // Print the local host name, address, port number
       << fServerAddress.GetHostName() 
       << "(port " << fServerAddress.GetPort() << ")" 
       << ", maxchild " << fMaxChild 
       << ", loglvl " << Msg::LevelAsString(fLogLevel)
       << ", maxinactive(sec) " << fMaxInactive  
       << ", currentchild " << fNumChild << "." << endl; 
  }
  else {
    ms << "PS: socket is not connected." << endl; 
  }

  return ms;

}

//_____________________________________________________________________________
void DDSParentServer::Run() {
  // Purpose: This is the main method of the DDSParentServer.  It
  //          listens for connections from new dispatcher clients.
  //          When a new connection with a client is established,
  //          the parent server will receive, process and respond to
  //          one and only one service request from the client before
  //          before disconnecting from that client.  This allows
  //          the parent server to remain open for listening to new connection
  //          requests. Each request received (of type DDS::EMessageType) 
  //          is responded to in return with a single message of type 
  //          DDS::EMessageType sent to the client upon completion of 
  //          servicing the request.  In this way the client can remain in 
  //          sync with the parent server's processing of its request. 
  //
  //          The services which the client can currently request are:
  //          DDS::kData  == Request for data.  This request spawns a separate
  //                         child server to handle the client.
  //          DDS::kShutdown    == Request to shutdown parent server.
  //                               Note that client must be local to
  //                               have authorization to shutdown parent
  //                               server.
  //          DDS:kStatus = Request for parent server status.  Returns 
  //                        kOk if alive.
  // 
  //          The return status with which the parent server can respond
  //          to the client may be one of the following:
  //          DDS::kPermissionDenied == Unauthorized client access denied.
  //          DDS::kMessageUnknown == Unrecognized message from client.
  //          DDS::kSystemError == Error occured in system call
  //                             (e.g. fork failed when spawning child server).
  //          DDS::kSocketError == An error return was received on the
  //                               socket receipt of the client's message.
  //          DDS::kSaturated == Maximum number of child servers reached.
  //          DDS::kOk == Parent server status ok or service completed ok.
  //
  // Arguments: none.
  //
  // Return: none.
  //
  // Contact: S. Kasahara
  //
  // Notes: The DDSParentServer will remain in the Run method until it 
  //        receives the DDS::kShutdown message from a local
  //        client.  
  //

   while ( !fShutdown ) {
    // Wait for new connection to parent server
    fTSocket = fTServerSocket -> Accept();
    if ( !fTSocket ) {
      MSG("DDS",Msg::kWarning) 
      << "PS: An error occured in TServerSocket::Accept while attempting to\n"
      << "accept a new client connection." << endl;
      continue; // loop and try again  
    }

    fClientAddress = fTSocket -> GetInetAddress();


    // Handle connected socket request
    if( ( fTSocket -> Recv(fMessageIn) ) > 0 ) {
      if ( fMessageIn -> What() != DDS::kStatus ) {
        MSG("DDS",Msg::kInfo) <<  "PS: Connected to new client at " 
                          << fClientAddress.GetHostName() << "/" 
                          << fClientAddress.GetHostAddress() 
                          << "(port " << fClientAddress.GetPort() << ")"
                          << endl;
      }
      fMessageOut.Reset();
      
      if ( !Authorize() ) {
        // Client is not authorized to access parent server
        fMessageOut.Reset(DDS::kPermissionDenied);
      }
      else {
        
        switch ( fMessageIn -> What()) {

        case DDS::kStatus:
          // Request for parentserver status
          Status();
          break;

        case DDS::kData:
          // Request for data
          fMessageOut.Reset(SpawnChildServer());
          break;

        case DDS::kShutdown:
          // Request to shutdown parent server
          fMessageOut.Reset(Shutdown());
          break;

        default:
          MSG("DDS",Msg::kWarning) << "PS: Unknown client message: " 
                               << fMessageIn->What() << " received.\n" 
                               <<"Unable to process client request." << endl;
          fMessageOut.Reset(DDS::kMessageUnknown);
          break;
        } // end of switch block

      } // end of fMessageIn processing

      delete fMessageIn; fMessageIn = 0;
    }
    else {
      MSG("DDS",Msg::kInfo) <<  "PS: Connected to new client at " 
                          << fClientAddress.GetHostName() << "/" 
                          << fClientAddress.GetHostAddress() 
                          << "(port " << fClientAddress.GetPort() << ")"
                          << endl;
      // Error return on TSocket::Recv call
      MSG("DDS",Msg::kWarning) 
      << "PS: Error returned from TSocket::Recv of client message.\n" << endl;
      fMessageOut.Reset(DDS::kSocketError);
    } // end of socket received fMessageIn block

    fTSocket -> Send(fMessageOut); // send return status to client
    // Client only allowed one chance to send valid message to parent server
    // to avoid backlog waiting for one client to get it right; child server
    // is more tolerant
    delete fTSocket; fTSocket=0;  // close and delete TSocket object 
    CheckChildStatus();   // check status of any exited children 
   
  } // while server processing block

}

//_____________________________________________________________________________
DDS::EMessageType DDSParentServer::Shutdown() {
  // Purpose: This Service shuts down the parent server.  This routine is 
  //          activated when the DDS::kShutdown message is received in
  //          DDSParentServer::Run().
  // 
  // Arguments: none.
  //
  // Return: DDS::EMessageType containing the return status of this service.
  //                           This service can return one of 2 messages:
  //                           DDS::kOk  (service provided ok)
  //                           DDS::kPermissionDenied (client doesn't have
  //                                permission to request shutdown)
  //
  // Contact: S. Kasahara
  //
  // Notes:  Client must be local (as determined by IsLocalClient()) to have 
  //         authority to shutdown parent server.

  DDS::EMessageType msgrc = DDS::kOk; // default return value

  if ( IsLocalClient() ) { 
    fShutdown = true;   // stops processing loop in Run
  }
  else {
    MSG("DDS",Msg::kWarning)
      << "PS: Nonlocal client at address:\n" << fClientAddress.GetHostName() 
      << "/" << fClientAddress.GetHostAddress() << "(port " 
      << fClientAddress.GetPort() << ")\n" 
      << "has requested shutdown of parent server on host "
      << gSystem->GetHostByName(gSystem->HostName()).GetHostName() 
      << ".\nPermission denied."<< endl;
    msgrc = DDS::kPermissionDenied; 
  }

  return msgrc;

}

//_____________________________________________________________________________
DDS::EMessageType DDSParentServer::Status() {
  // Purpose: This Service returns the parent server status.  This routine is 
  //          activated when the DDS::kStatus message is received in
  //          DDSParentServer::Run().
  // 
  // Arguments: none.
  //
  // Return: DDS::EMessageType containing the return status of this service.
  //                           This service can return one  message:
  //                           DDS::kOk  (service provided ok)
  //
  // Contact: S. Kasahara
  //
  // Notes: Eventually could send status object back to client. Currently
  //        always returns okay (e.g. this server is up and listening).

  DDS::EMessageType msgrc = DDS::kOk; // default return value
  CheckChildStatus();  // first check for any newly returned children
  if( fNumChild >= fMaxChild ) {
    msgrc = DDS::kSaturated;
  }
  fMessageOut.Reset(msgrc);
  fMessageOut.WriteObject(&fPSStatus);
  
  return msgrc;

}

//_____________________________________________________________________________
DDS::EMessageType DDSParentServer::SpawnChildServer() {
  // Purpose: This Service spawns a child server to handle client request 
  //          for near-online data. This routine is activated when the 
  //          DDS:kData message is received in DDSParentServer::Run(). 
  // 
  // Arguments: none.
  //
  // Return: DDS::EMessageType containing the return status of this 
  //                           service.  This service can return one of
  //                           3 messages: 
  //                           DDS::kOk (service provided ok)
  //                           DDS::kSaturated (max no. of children reached)
  //                           DDS::kSystemError (fork or exec failed)
  //
  // Contact: S. Kasahara
  //
  // Notes: Local client child servers are executed at a higher priority 
  //        than remote clients.  A successfully spawned child increments 
  //        the fNumChild counter by 1.
  //        
  
  DDS::EMessageType msgrc = DDS::kOk; // default return value 

  DDSClientId* ddsId = dynamic_cast<DDSClientId*>
                      (fMessageIn -> ReadObject(fMessageIn->GetClass()));
  
  if ( !ddsId ) {
     MSG("DDS",Msg::kWarning) 
     << "PS: SpawnChildServer failed to received DDSClientId from client."
     << "\nClient sw likely out-of-date. Refuse to spawn childserver." << endl;
     msgrc = DDS::kError;
     return msgrc;
  }

  // First check to see if parent has reached the saturation level with
  // the number of supported children.
  CheckChildStatus();  // first check for any newly returned children
  if( fNumChild == fMaxChild ) {
    MSG("DDS",Msg::kWarning) << "PS: Unable to spawn new child per request\n"
    << "because number of child servers running (" << fNumChild 
    << ") is maximum allowed." << endl;
    msgrc = DDS::kSaturated;
  }
  else {
    // Fork a child process to handle the new client
    int pid;
    if( (pid = fork()) < 0) {
      // Error in fork, unable to spawn child
      MSG("DDS",Msg::kError) 
      << "PS: Unable to spawn child.  Fork failed with error "
      << strerror(errno) << endl;
      msgrc = DDS::kSystemError;
    }
    else if ( pid > 0 ) {
      // Successful fork, report new child process id
      MSG("DDS",Msg::kInfo) << "PS: Spawned child with process id " << pid 
                            << " to handle client request." << endl;
      fNumChild++;

      // However, wait one second to check to see if execlp failed to start
      // ChildServer properly. This is done so that the correct status can be
      // reported back to the client if the execlp failed.
      gSystem -> Sleep(1000);
      int childpid;  // pid number of exited child processes
      Int_t childstat; // exit status of child process
      childpid = waitpid(pid, &childstat, WNOHANG);
      if (childpid > 0) {
        // Spawned child process exited prematurely
        fNumChild--; 
        msgrc = DDS::kSystemError;
      }
      else {
        // Successfully spawned child. Enter pid in ClientId
        ddsId -> Connected(pid);
        fPSStatus.SetClientId(pid,ddsId);
        MSG("DDS",Msg::kInfo) << fPSStatus << endl;
      }
    }

    if (pid == 0) {
      // In the child process.
      // Must Close serversocket in child because reference count of
      // serversocket is incremented by 1 when child is created. Close
      // just decrements reference count of serversocket by 1 (child won't
      // use server socket).  
      fTServerSocket -> Close();
      Int_t sockfd = fTSocket->GetDescriptor(); //connected socket descriptor
      char csockfd[20];
      sprintf(csockfd,"%i",sockfd); // convert to character string
      // Determine the priority level at which the child server will run.
      // Local clients are served at a priority level higher than remote 
      // clients. 
      Int_t niceincr = (IsLocalClient()) ? 0 : 5;
      char cniceincr[20];
      sprintf(cniceincr,"%i",niceincr); // convert to character string
      char cloglevel[20];
      sprintf(cloglevel,"%i",fLogLevel); // convert loglevel to char string
      char cmaxinactive[20];
      sprintf(cmaxinactive,"%i",fMaxInactive); // maxinactive to char string
      // Pass the program which handles the child the connected socket 
      // descriptor
      execlp("ddschildserver","ddschildserver", csockfd, cniceincr,cloglevel,
             cmaxinactive,(char *)0);
      // Only reach this point if execlp fails
      MSG("DDS",Msg::kError) << "PS: execlp of ChildServer\n" 
       << "failed with error: " << strerror(errno) << ".  Please check that\n"
   <<"the location of the ChildServer is in the calling process search PATH.\n"
     << endl;
      exit(1);
    } // end of child processing

  } 

  return msgrc;

}


  

---