JiveXML::JiveXMLServer Class Reference

#include <JiveXMLServer.h>

Inheritance diagram for JiveXML::JiveXMLServer:

Collaboration diagram for JiveXML::JiveXMLServer:

Public Member Functions
	JiveXMLServer (int port=0)
	Constructor.
virtual	~JiveXMLServer ()
	Destructor.
IEventReceiver methods
virtual StatusCode	UpdateEventForStream (const EventStreamID &evtStreamID, const std::string &event) override
	Put this event as new current event for stream given by name.
IEventServer methods
virtual std::vector< std::string >	GetStreamNames () const override
	get the names of all the streams
virtual const EventStreamID	GetEventStreamID (const std::string &streamName) const override
	get the current EventStreamID for a particular stream
virtual const std::string	GetEvent (const EventStreamID &evtStreamID) const override
	get the current event for a particular stream
virtual int	GetState () const override
	get the Status of the application
virtual void	Message (const MSG::Level level, const std::string &msg) const override
	This function is exposed to allow using ERS messaging service from other threads.
virtual MSG::Level	LogLevel () const override
	Get the logging level.

Event serving thread control
static std::promise< int > m_receivedSignal	ATLAS_THREAD_SAFE
int	m_portNumber
EventStreamMap	m_eventStreamMap
pthread_mutex_t m_accessLock	ATLAS_THREAD_SAFE
pthread_t	m_ServerThreadHandle
bool	m_runServerThread
StatusCode	StartServingThread ()
	Start the serving thread.
StatusCode	StopServingThread ()
	Stop the serving thread.
virtual bool	GetRunServerFlag () const override
	The server thread will stop once this flag is set to false.
virtual void	ServerThreadStopped () override
	Callback whenever the server thread is stopped.
void	Wait ()
	Wait for the server finish.
static void	signalHandler (int signum)
	When the signal handler is called, switch the lock to the post condition.

Detailed Description

Definition at line 36 of file JiveXMLServer.h.

Constructor & Destructor Documentation

JiveXMLServer()

JiveXMLServer::JiveXMLServer

(

int

port = 0

)

Constructor.

Definition at line 41 of file JiveXMLServer.cxx.

                                        :
    m_portNumber(port){
 
    //Make sure ServerThread does not start unexpectedly
    m_runServerThread = false ;
 
    //And then start it
    StartServingThread().ignore();
 
    //Also register the signal handlers
    signal( SIGINT , JiveXMLServer::signalHandler );
    signal( SIGTERM, JiveXMLServer::signalHandler );
  }

~JiveXMLServer()

JiveXMLServer::~JiveXMLServer ( )

virtual

Destructor.

Definition at line 58 of file JiveXMLServer.cxx.

                               {
  
    //Just  stop the serving thread
    StopServingThread().ignore();
  }

Member Function Documentation

GetEvent()

const std::string JiveXMLServer::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 363 of file JiveXMLServer.cxx.

                                                                                  {
    
    //Obtain an exclusive access lock
    int retVal = pthread_mutex_lock(&m_accessLock);
    if ( retVal != 0 ){
      ERS_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);
 
    //Initialize with an empty event stream
    std::string event;
 
    //If the element is found, get a copy of the found event string
    if ( MapItr != m_eventStreamMap.end()){
      event = std::string((*MapItr).second);
    }
 
    //Release the lock
    retVal = pthread_mutex_unlock(&m_accessLock);
    if ( retVal != 0 )
      ERS_ERROR("Unable to release access lock after getting stream event: " << fmterror(retVal));
 
    return event;
  }

GetEventStreamID()

const EventStreamID JiveXMLServer::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 331 of file JiveXMLServer.cxx.

                                                                                        {
    
    //Obtain an exclusive access lock
    int retVal = pthread_mutex_lock(&m_accessLock);
    if ( retVal != 0 ){
      ERS_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));
 
    //Initialize with an invalid event stream identifier
    EventStreamID streamID = EventStreamID("");
 
    //If the element is found, get a copy of the found event stream identifier
    if ( MapItr != m_eventStreamMap.end()){
      streamID = EventStreamID((*MapItr).first);
    }
 
    //Release the lock
    retVal = pthread_mutex_unlock(&m_accessLock);
    if ( retVal != 0 )
      ERS_ERROR("Unable to release access lock after getting stream ID: " << fmterror(retVal));
 
    return streamID;
 
  }

GetRunServerFlag()

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

inlineoverridevirtual

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

Implements JiveXML::IEventServer.

Definition at line 78 of file JiveXMLServer.h.

{ return m_runServerThread; };

GetState()

int JiveXMLServer::GetState ( ) const

overridevirtual

get the Status of the application

Return the current athena state.

at the moement return a fixed value - will change this to some timeout condition soon.

Implements JiveXML::IEventServer.

Definition at line 285 of file JiveXMLServer.cxx.

                                    {
    return 3;
  }

GetStreamNames()

std::vector< std::string > JiveXMLServer::GetStreamNames ( ) const

overridevirtual

get the names of all the streams

Return an array with all the stream names.

Implements JiveXML::IEventServer.

Definition at line 296 of file JiveXMLServer.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 ){
      ERS_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 )
      ERS_ERROR("Unable to release access lock after getting stream names: " << fmterror(retVal));
 
    //Return the list of names
    return StreamNames;
  }

LogLevel()

MSG::Level JiveXMLServer::LogLevel ( ) const

overridevirtual

Get the logging level.

Currently only used to suppress client hostname lookup if not in debug mode.

Implements JiveXML::IMessage.

Definition at line 407 of file JiveXMLServer.cxx.

                                          {
    //set to fixed value for now
    return MSG::DEBUG;
  }

Message()

void JiveXMLServer::Message ( const MSG::Level level, const std::string & msg ) const

overridevirtual

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

Deliver a message - possibly from another thread - to ERS;.

Implements JiveXML::IMessage.

Definition at line 395 of file JiveXMLServer.cxx.

                                                                               {
    //Deliver message to the proper stream
    if (level <= MSG::DEBUG) ERS_REPORT_IMPL( ers::debug, ers::Message, msg, level);
    if (level == MSG::INFO) ERS_REPORT_IMPL( ers::info, ers::Message, msg, );
    if (level == MSG::WARNING) ERS_REPORT_IMPL( ers::warning, ers::Message, msg, );
    if (level == MSG::ERROR) ERS_REPORT_IMPL( ers::error, ers::Message, msg, );
    if (level >= MSG::FATAL) ERS_REPORT_IMPL( ers::fatal, ers::Message, msg, );
  }

ServerThreadStopped()

void JiveXMLServer::ServerThreadStopped ( )

overridevirtual

Callback whenever the server thread is stopped.

When the server thread stopped, we will also call the the signal handler with the special value -1.

Implements JiveXML::IServer.

Definition at line 198 of file JiveXMLServer.cxx.

                                         {
    //call the signal handler, so we will also reach post condition
    signalHandler(-1);
  }

signalHandler()

void JiveXMLServer::signalHandler ( int signum )

staticprivate

When the signal handler is called, switch the lock to the post condition.

Definition at line 176 of file JiveXMLServer.cxx.

                                             {
    //Store signal and notify thread
    m_receivedSignal.set_value(signal);
  }

StartServingThread()

StatusCode JiveXMLServer::StartServingThread

(

)

Start the serving thread.

Create the server by.

• passing a this-pointer as an argument
• setting the thread-running flag
• starting the server thread

Definition at line 70 of file JiveXMLServer.cxx.

                                              {
 
    ERS_DEBUG(MSG::VERBOSE,"StartServingThread()");
 
    //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){
       ERS_WARNING("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
      ERS_WARNING("Thread creation failed");
      return StatusCode::FAILURE;
    }
 
    return StatusCode::SUCCESS;
 
  }

StopServingThread()

StatusCode JiveXMLServer::StopServingThread

(

)

Stop the serving 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 107 of file JiveXMLServer.cxx.

                                             {
 
    ERS_DEBUG(MSG::VERBOSE,"StopServingThread()");

 
    //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){
      ERS_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 ; 
 
    // Now issue the call with a timeout
    struct timeval timeout; timeout.tv_sec = 1; timeout.tv_usec = 0;
// 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
#if defined(__clang__) && __clang_major__ >= 19
# pragma clang diagnostic push
# pragma clang diagnostic ignored "-Wcast-function-type-mismatch"
#endif
    clnt_call(client, NULLPROC, (xdrproc_t)xdr_void, NULL, (xdrproc_t)xdr_void, NULL, timeout);
#if defined(__clang__) && __clang_major__ >= 19
# pragma clang diagnostic pop
#endif
#if __GNUC__ >= 8
# pragma GCC diagnostic pop
#endif
 
    // A pointer to the return value of the thread
    void* ret = NULL;
    // wait till the server thread has finished
    ERS_INFO("Waiting for server thread to terminate ...");
    pthread_join(m_ServerThreadHandle, &ret);
    ERS_INFO(" ... finished server thread");
 
    //check if there was a return value
    if (ret){
      //Get the return value
      unsigned long NRequests = *(unsigned long*)ret;
      ERS_DEBUG(MSG::DEBUG,"Server thread stopped after handling " << NRequests << " requests");
    } else 
      ERS_WARNING("Server thread stopped unexpectedly");
 
    //Destroy the access lock
    int retVal = pthread_mutex_destroy(&m_accessLock);
    if (retVal != 0){
       ERS_WARNING("Unable to destroy access lock after stopping server: " << fmterror(retVal));
       return StatusCode::FAILURE;
    }
 
    return StatusCode::SUCCESS;
  }

UpdateEventForStream()

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

overridevirtual

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

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 207 of file JiveXMLServer.cxx.

                                                                                                           {
 
    ERS_DEBUG(MSG::VERBOSE,"UpdateEventForStream");
 
    //Check that the event stream id is valid
    if (!evtStreamID.isValid()){
      ERS_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 ){
      ERS_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 ){
      ERS_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 30 second and 0 nanoseconds
    struct timespec timeout = { 30, 0 };
    //Try to obtain the lock
    int retVal = pthread_mutex_timedlock(&m_accessLock, &timeout);
    if ( retVal != 0 ){
      ERS_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 ) {
      ERS_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 ){
      ERS_ERROR("Unable to release access lock after updating event: " << fmterror(retVal));
      return StatusCode::FAILURE;
    }
 
    ERS_DEBUG(MSG::DEBUG, "Updated stream " << evtStreamID.StreamName() 
               << " with event Nr. " << evtStreamID.EventNumber() 
               << " from run Nr. " << evtStreamID.RunNumber());
 
    return StatusCode::SUCCESS;
  }

Wait()

void JiveXMLServer::Wait

(

)

Wait for the server finish.

Wait for the server thread to finish.

This can be because a) we received a SIGTERM or SIGINT signal b) the server thread stopped by itself

Definition at line 186 of file JiveXMLServer.cxx.

                          {
    auto signal = m_receivedSignal.get_future();
    //just wait for a signal
    signal.wait();
    //Tell why the lock was released
    ERS_INFO("Reached post-condition after received signal " << signal.get() );
  }

Member Data Documentation

ATLAS_THREAD_SAFE [1/2]

std::promise<int> m_receivedSignal JiveXML::JiveXMLServer::ATLAS_THREAD_SAFE

inlinestaticprivate

Definition at line 95 of file JiveXMLServer.h.

ATLAS_THREAD_SAFE [2/2]

pthread_mutex_t m_accessLock JiveXML::JiveXMLServer::ATLAS_THREAD_SAFE

mutableprivate

Definition at line 105 of file JiveXMLServer.h.

m_eventStreamMap

EventStreamMap JiveXML::JiveXMLServer::m_eventStreamMap

private

Definition at line 102 of file JiveXMLServer.h.

m_portNumber

int JiveXML::JiveXMLServer::m_portNumber

private

Definition at line 99 of file JiveXMLServer.h.

m_runServerThread

bool JiveXML::JiveXMLServer::m_runServerThread

private

Definition at line 112 of file JiveXMLServer.h.

m_ServerThreadHandle

pthread_t JiveXML::JiveXMLServer::m_ServerThreadHandle

private

Definition at line 108 of file JiveXMLServer.h.

---

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

• JiveXMLServer.h
• JiveXMLServer.cxx