JiveXML::ONCRPCServerSvc Class Reference

This athena service will create an ONC/RPC (aka SunRPC) server, that can provide the athena status as well as data strings from events. More...

#include <ONCRPCServerSvc.h>

Inheritance diagram for JiveXML::ONCRPCServerSvc:

Collaboration diagram for JiveXML::ONCRPCServerSvc:

Public Member Functions
	ONCRPCServerSvc (const std::string &name, ISvcLocator *sl)
	Default constructor. More...
virtual	~ONCRPCServerSvc ()
	Destructor. More...
virtual StatusCode	initialize () override
	Gaudi default methods. More...
virtual StatusCode	finalize () override
	Finalize - called once at the end. More...

Server methods
int	m_portNumber
EventStreamMap	m_eventStreamMap
pthread_mutex_t m_accessLock	ATLAS_THREAD_SAFE
pthread_t	m_ServerThreadHandle
bool	m_runServerThread
virtual int	GetState () const override
	get the Status of the application More...
virtual std::vector< std::string >	GetStreamNames () const override
	get the names of all the streams More...
virtual const EventStreamID	GetEventStreamID (const std::string &streamName) const override
	get the current EventStreamID for a particular stream More...
virtual const std::string	GetEvent (const EventStreamID &evtStreamID) const override
	get the current event for a particular stream More...
virtual StatusCode	UpdateEventForStream (const EventStreamID &evtStreamID, const std::string &event) override
	Put this event as new current event for stream given by name. More...
virtual void	Message (const MSG::Level level, const std::string &mesg) const override
	This function is exposed to allow using athena messaging service from other threads. More...
virtual MSG::Level	LogLevel () const override
	Get the logging level. More...
virtual bool	GetRunServerFlag () const override
	The server thread will stop once this flag is set to false. More...
StatusCode	StartServer ()
	Start the server thread. More...
StatusCode	StopServer ()
	Stop the server thread. More...
virtual void	ServerThreadStopped () override
	Callback when server thread terminates. More...

Detailed Description

This athena service will create an ONC/RPC (aka SunRPC) server, that can provide the athena status as well as data strings from events.

It can handle several different event streams, but for each stream only the latest event is available. This service is usually used by the Athena JiveXML algorithm providing XML events for an Atlantis Java client.

The server will create a single server thread, which loops over the sockets until the m_runServerThread flag is set to false. On each request, it will call the ONCRPCRequestHandler, which will generate a new thread for each request. The actual response to the requests happens in these ONCRPCDispatchThread. A list of the thread handles is kept in the a ThreadCollection, where each thread will add and remove itself. In the finalization stage, the server will wait for all dispatch threats to finish before exiting.

On all stages, the server thread will have a handle to the athena service via the thread specific key ServerSvcKey. It is thus possible to have many servers running simultaneously. Note however, that currently RPC will forbid you to start several servers with the same ONCRPCSERVERPROG.

Definition at line 42 of file ONCRPCServerSvc.h.

Constructor & Destructor Documentation

ONCRPCServerSvc()

JiveXML::ONCRPCServerSvc::ONCRPCServerSvc	(	const std::string &	name,
		ISvcLocator *	sl
	)

Default constructor.

Constructor.

Definition at line 268 of file ONCRPCServerSvc.cxx.

                                                                          :
    base_class( name, sl )
  {
      //declare port number property
      declareProperty("PortNumber",m_portNumber=0,"Port number to bind - assigned dynamically if set to zero [default: 0]");
 
      //Make sure ServerThread does not start unexpectedly
      m_runServerThread = false ;
 
      ATH_MSG_VERBOSE( "Output level is set to " << (int)msg().level()  );
 
  }

~ONCRPCServerSvc()

JiveXML::ONCRPCServerSvc::~ONCRPCServerSvc ( )

virtual

Destructor.

Definition at line 284 of file ONCRPCServerSvc.cxx.

                                   {
 
    ATH_MSG_VERBOSE( "Destructor() "  );
  }

Member Function Documentation

finalize()

StatusCode JiveXML::ONCRPCServerSvc::finalize ( )

overridevirtual

Finalize - called once at the end.

• shut down server

Definition at line 314 of file ONCRPCServerSvc.cxx.

                                       {
 
    ATH_MSG_VERBOSE( "Finalize()"  );
 
    //Stop the server thread and return status code
    return StopServer();
  }

GetEvent()

const std::string JiveXML::ONCRPCServerSvc::GetEvent ( const EventStreamID &  evtStreamID ) const

overridevirtual

get the current event for a particular stream

Return the event for a given stream.

Implements JiveXML::IEventServer.

Definition at line 241 of file ONCRPCServerSvc.cxx.

                                                                                    {
    
    //Obtain an exclusive access lock
    int retVal = pthread_mutex_lock(&m_accessLock);
    if ( retVal != 0 ){
      ATH_MSG_ERROR( "Unable to obtain access lock to get event: " << fmterror(retVal)  );
      return std::string("");
    }
 
    // Search the entry in the map
    EventStreamMap::const_iterator MapItr = m_eventStreamMap.find(evtStreamID);
 
    //Release the lock
    retVal = pthread_mutex_unlock(&m_accessLock);
    if ( retVal != 0 )
      ATH_MSG_ERROR( "Unable to release access lock after getting stream event: " << fmterror(retVal)  );
 
    //If the element was not found return an empty string
    if ( MapItr == m_eventStreamMap.end()) return std::string("");
 
    //Return a copy of the found event string
    return std::string((*MapItr).second);
  }

GetEventStreamID()

const EventStreamID JiveXML::ONCRPCServerSvc::GetEventStreamID ( const std::string &  streamName ) const

overridevirtual

get the current EventStreamID for a particular stream

Return the EventStreamID for the last event of a given stream.

Implements JiveXML::IEventServer.

Definition at line 214 of file ONCRPCServerSvc.cxx.

                                                                                          {
    
    //Obtain an exclusive access lock
    int retVal = pthread_mutex_lock(&m_accessLock);
    if ( retVal != 0 ){
      ATH_MSG_ERROR( "Unable to obtain access lock to get stream ID: " << fmterror(retVal)  );
      return EventStreamID("");
    }
 
    // Search the entry in the map
    EventStreamMap::const_iterator MapItr = m_eventStreamMap.find(EventStreamID(StreamName));
 
    //Release the lock
    retVal = pthread_mutex_unlock(&m_accessLock);
    if ( retVal != 0 )
      ATH_MSG_ERROR( "Unable to release access lock after getting stream ID: " << fmterror(retVal)  );
 
    //If the element was not found return an invalid ID
    if ( MapItr == m_eventStreamMap.end()) return EventStreamID("");
 
    //Return the found event stream identifier
    return (*MapItr).first;
  }

GetRunServerFlag()

virtual bool JiveXML::ONCRPCServerSvc::GetRunServerFlag ( ) const

inlineoverridevirtual

The server thread will stop once this flag is set to false.

Implements JiveXML::IEventServer.

Definition at line 82 of file ONCRPCServerSvc.h.

{ return m_runServerThread; };

GetState()

int JiveXML::ONCRPCServerSvc::GetState ( ) const

overridevirtual

get the Status of the application

Return the current athena state.

at the moment simply return state machine state - we might consider checking for state transitions (m_state != m_targetState) and return a "busy" value. At the moment, all state transitions should be instantanious, so this is not necessary

Implements JiveXML::IEventServer.

Definition at line 166 of file ONCRPCServerSvc.cxx.

                                      {
    return (int)FSMState();
  }

GetStreamNames()

std::vector< std::string > JiveXML::ONCRPCServerSvc::GetStreamNames ( ) const

overridevirtual

get the names of all the streams

Return an array with all the stream names.

Implements JiveXML::IEventServer.

Definition at line 179 of file ONCRPCServerSvc.cxx.

                                                             {
    
    //Create a vector that can be returned
    std::vector<std::string> StreamNames;
    
    //Obtain an exclusive access lock
    int retVal = pthread_mutex_lock(&m_accessLock);
    if ( retVal != 0 ){
      ATH_MSG_ERROR( "Unable to obtain access lock to get stream names: " << fmterror(retVal)  );
      return StreamNames;
    }
 
    // Iterate over map to get entries
    EventStreamMap::const_iterator MapItr = m_eventStreamMap.begin();
    for ( ; MapItr != m_eventStreamMap.end(); ++MapItr){
      
      //Get the EventStreamID object
      EventStreamID EvtStrID = (*MapItr).first;
 
      //Add the name of this EventStreamID to the list of stream names
      StreamNames.push_back(EvtStrID.StreamName());
    }
 
    //Release the lock
    retVal = pthread_mutex_unlock(&m_accessLock);
    if ( retVal != 0 )
      ATH_MSG_ERROR( "Unable to release access lock after getting stream names: " << fmterror(retVal)  );
 
    //Return the list of names
    return StreamNames;
  }

initialize()

StatusCode JiveXML::ONCRPCServerSvc::initialize ( )

overridevirtual

Gaudi default methods.

Initialize - called once in beginning.

• create server

Create the server itself

Definition at line 293 of file ONCRPCServerSvc.cxx.

                                         {
 
    //Initialize Service base
    if (Service::initialize().isFailure()) return StatusCode::FAILURE;
    
     //Initialize message stream level
    msg().setLevel(outputLevel());
    ATH_MSG_VERBOSE( "Initialize()"  );
    ATH_MSG_VERBOSE( "Output level is set to " << (int)msg().level()  );
 
    ATH_CHECK( StartServer() );
    return StatusCode::SUCCESS;
  }

LogLevel()

MSG::Level JiveXML::ONCRPCServerSvc::LogLevel ( ) const

overridevirtual

Get the logging level.

Return the logging level - possibly to detect whether it is worth retrieving the information to be logged.

Implements JiveXML::IMessage.

Definition at line 157 of file ONCRPCServerSvc.cxx.

                                           {
    //return the current logging level
    return msg().level();
  }

Message()

void JiveXML::ONCRPCServerSvc::Message ( const MSG::Level  level, const std::string &  mesg  ) const

overridevirtual

This function is exposed to allow using athena messaging service from other threads.

Deliver a message - possibly from another thread - to the athena msgSvc;.

Implements JiveXML::IMessage.

Definition at line 144 of file ONCRPCServerSvc.cxx.

  {
    //Deliver message
    msg() << level << mesg << endmsg;
    //Spit it out immediately
    msg().flush();
  }

ServerThreadStopped()

void JiveXML::ONCRPCServerSvc::ServerThreadStopped ( )

overridevirtual

Callback when server thread terminates.

This callback will get called whenever the server thread stops.

Implements JiveXML::IServer.

Definition at line 130 of file ONCRPCServerSvc.cxx.

                                           {
    //Check if the thread stopped while the run server flag was set
    if (m_runServerThread){
      //Deliver an error message
      ATH_MSG_ERROR( "Server thread stopped while run-server-flag is set!"  );
      //since we can't return statusCode failure (we're in a different thread!)
      //set the flag to mark the server as dead
      m_runServerThread = false;
    }
  }

StartServer()

StatusCode JiveXML::ONCRPCServerSvc::StartServer

(

)

Start the server thread.

Create the server by.

• passing a set of arguments including a this-pointer
• setting the thread-running flag
• starting the server thread

Definition at line 33 of file ONCRPCServerSvc.cxx.

                                         {
 
    //Initialize access lock mechanism to ensure that the data map is never
    //accessed by more than one thread at the same time. NULL means default
    int retVal = pthread_mutex_init(&m_accessLock,NULL);
    if (retVal != 0){
      ATH_MSG_ERROR( "Unable to initialize access lock while starting server: " << fmterror(retVal)  );
       return StatusCode::FAILURE;
    }
 
    //The arguments passed on to the server - create new object on the heap that
    //is persistent through the lifetime of the thread
    ServerThreadArguments* args = new ServerThreadArguments(this, m_portNumber);
 
    //set runServer flag to true, so the  thread will start
    m_runServerThread = true ;
 
    //create thread itself
    if ( ( pthread_create(&m_ServerThreadHandle, NULL , &JiveXML::ONCRPCServerThread, (void *) args )) != 0){
      
      //Create thread failed
      ATH_MSG_ERROR( "Thread creation failed"  );
      return StatusCode::FAILURE;
    }
 
    return StatusCode::SUCCESS;
 
  }

StopServer()

StatusCode JiveXML::ONCRPCServerSvc::StopServer

(

)

Stop the server thread.

Stop the server by.

• unsetting the thread-running flag
• waiting till the server thread has finished
• destroying the thread handle

Ping the server which will cause another request Otherwise the server won't update its loop condition

Definition at line 68 of file ONCRPCServerSvc.cxx.

                                        {
 
    ATH_MSG_VERBOSE( "StopServer()"  );
 
    //Create a client - this will already cause an update of the file
    //descriptors on the sockets
    CLIENT* client = clnt_create("localhost", ONCRPCSERVERPROG,ONCRPCSERVERVERS, "tcp");
    if (!client){
      ATH_MSG_ERROR( "Unable to create shutdown client for local server"  );
      return StatusCode::FAILURE;
    }
 
    // Next unset the runServerThread flag, which will cause the loop to stop
    // This needs to happen after client creation, otherwise the next call won't
    // be serverd anymore
    m_runServerThread = false ; 
 
// xdr_void is defined inconsistently in xdr.h and gets a warning from gcc8.
#if __GNUC__ >= 8
# pragma GCC diagnostic push
# pragma GCC diagnostic ignored "-Wcast-function-type"
#endif
    // Now issue the call with a timeout
    struct timeval timeout; timeout.tv_sec = 1; timeout.tv_usec = 0;
    clnt_call(client, NULLPROC, (xdrproc_t)xdr_void, NULL, (xdrproc_t)xdr_void, NULL, timeout);
#if __GNUC__ >= 8
# pragma GCC diagnostic pop
#endif
 
    // A pointer to the return value of the thread
    void* ret;
    // wait till the server thread has finished
    ATH_MSG_INFO( "Waiting for server thread to terminate ..."  );
    pthread_join(m_ServerThreadHandle, &ret);
    ATH_MSG_INFO( " ... finished server thread"  );
 
    //check if there was a return value
    if (ret){
      //Get the return value
      unsigned long NRequests = *(unsigned long*)ret;
      ATH_MSG_DEBUG( "Server thread stopped after handling " << NRequests << " requests"  );
    } else 
      ATH_MSG_WARNING( "Server thread had stopped unexpectedly"  );
 
    //Destroy the access lock
    int retVal = pthread_mutex_destroy(&m_accessLock);
    if (retVal != 0){
       ATH_MSG_ERROR( "Unable to destroy access lock after stopping server: " << fmterror(retVal)  );
       return StatusCode::FAILURE;
    }
 
    return StatusCode::SUCCESS;
  }

UpdateEventForStream()

StatusCode JiveXML::ONCRPCServerSvc::UpdateEventForStream ( const EventStreamID &  evtStreamID, const std::string &  event  )

overridevirtual

Put this event as new current event for stream given by name.

Implementation of IServerSvc method.

Get one event and put it as the new event for the this stream, which is identified by EventStreamID

Implements JiveXML::IEventReceiver.

Definition at line 326 of file ONCRPCServerSvc.cxx.

                                                                                                             {
 
    ATH_MSG_VERBOSE( "UpdateEventForStream()"  );
 
    //Make sure the server is still running
    if (! m_runServerThread ){
      ATH_MSG_ERROR( "Server thread is not running - refusing to update event"  );
      return StatusCode::FAILURE;
    }
 
    //Check that the event stream id is valid
    if (!evtStreamID.isValid()){
      ATH_MSG_ERROR( "Invalid event stream identifier - cannot add event"  );
      return StatusCode::FAILURE;
    }
 
    //Make sure we don't have already exceeded the maximum number of streams
    if (m_eventStreamMap.size() > NSTREAMMAX ){
      ATH_MSG_ERROR( "Reached max. allowed number of streams " << NSTREAMMAX << " - cannot add event"  );
      return StatusCode::FAILURE;
    }
 
    //Make sure the event is not larger than the allowed maximal size
    if (event.length() > NBYTESMAX ){
      ATH_MSG_ERROR( "Event is larger than allowed max. of " << NBYTESMAX << " bytes - cannot add event"  );
      return StatusCode::FAILURE;
    }
 
    //Make sure we are the only one accessing the data right now, by trying to
    //obtain a lock. If the lock can not be obtained after a certain time, an
    //error is reported
    
    //Timeout of 5 second and 0 nanoseconds
    struct timespec timeout = { 5, 0 };
    //Try to obtain the lock
#ifndef __APPLE__
    int retVal = pthread_mutex_timedlock(&m_accessLock, &timeout);
#else
    int retVal = pthread_mutex_lock(&m_accessLock);
#endif
    if ( retVal != 0 ){
      ATH_MSG_ERROR( "Unable to obtain access lock to update event: " << fmterror(retVal)  );
      return StatusCode::FAILURE;
    }
 
    //Using try/catch to ensure the mutex gets unlocked in any case
    try {
 
      //Using std::map::operator[] and std::map::insert() will create a new event
      //if it did not exist, otherwise just replace the existing entry (making a
      //copy of the std::string) but would not update the key which holds new
      //event/run number. Therefore delete existing entry first.
 
      //Delete old entry if there is one
      EventStreamMap::iterator OldEvtItr = m_eventStreamMap.find(evtStreamID);
      if (OldEvtItr != m_eventStreamMap.end()) 
        m_eventStreamMap.erase(OldEvtItr);
      
      //Now add the new event
      m_eventStreamMap.insert(EventStreamPair(evtStreamID,event));
    
    } catch ( const std::exception& e ) {
      ATH_MSG_ERROR( "Exception caught while updating event for stream " << evtStreamID.StreamName()
                     << ": " << e.what()  );
      //Also release the lock in this case
      pthread_mutex_unlock(&m_accessLock);
      //before we return
      return StatusCode::FAILURE;
    }
 
    //Finally release the lock again
    retVal = pthread_mutex_unlock(&m_accessLock);
    if ( retVal != 0 ){
      ATH_MSG_ERROR( "Unable to release access lock after updating event: " << fmterror(retVal)  );
      return StatusCode::FAILURE;
    }
 
    ATH_MSG_DEBUG( "Updated stream " << evtStreamID.StreamName() 
                   << " with event Nr. " << evtStreamID.EventNumber() 
                   << " from run Nr. " << evtStreamID.RunNumber()  );
 
    return StatusCode::SUCCESS;
  }

Member Data Documentation

ATLAS_THREAD_SAFE

pthread_mutex_t m_accessLock JiveXML::ONCRPCServerSvc::ATLAS_THREAD_SAFE

mutableprivate

Definition at line 100 of file ONCRPCServerSvc.h.

m_eventStreamMap

EventStreamMap JiveXML::ONCRPCServerSvc::m_eventStreamMap

private

Definition at line 97 of file ONCRPCServerSvc.h.

m_portNumber

int JiveXML::ONCRPCServerSvc::m_portNumber

private

Definition at line 94 of file ONCRPCServerSvc.h.

m_runServerThread

bool JiveXML::ONCRPCServerSvc::m_runServerThread

private

Definition at line 107 of file ONCRPCServerSvc.h.

m_ServerThreadHandle

pthread_t JiveXML::ONCRPCServerSvc::m_ServerThreadHandle

private

Definition at line 103 of file ONCRPCServerSvc.h.

---

The documentation for this class was generated from the following files:

• ONCRPCServerSvc.h
• ONCRPCServerSvc.cxx