JiveXML Namespace Reference

This header is shared inbetween the C-style server thread and the C++ Athena ServerSvc. More...

Namespaces
	MuonFullIDHelper
	SiSpacePointRetrieverHelpers
	TrackParticleRetrieverHelpers
	TrackRetrieverHelpers
	Namespace for all helper functions.

Classes
class	AlgoJiveXML
	Converts Reconstructed Event Data objects into XML files which can be read by, for example, the Atlantis graphics package. More...
class	AODCaloClusterRetriever
	Retrieves all Calo Cluster objects. More...
class	AODJetRetriever
	Retrieves all Jet objects (JetEvent/Jet) More...
class	BadLArRetriever
	Retrieves all Calo Cluster objects. More...
class	BadTileRetriever
	Retrieves all Tile Calo Cell objects. More...
class	BeamSpotRetriever
	Retrieve the BeamSpot information (official ID s/w result) More...
class	BJetRetriever
	Retrieves Jet objects, filter from weight for BJet (JetEvent/Jet) More...
class	CaloClusterRetriever
	Retrieves all Calo Cluster objects. More...
class	CaloFCalRetriever
	Retrieves all Calo Cluster objects. More...
class	CaloHECRetriever
	Retrieves all Calo Cluster objects. More...
class	CaloLArRetriever
	Retrieves all Calo Cluster objects. More...
class	CaloMBTSRetriever
	Retrieves all Calo Cluster objects. More...
class	CaloTileRetriever
	Retrieves all Tile Calo Cell objects. More...
class	CompositeParticleRetriever
	Retrieves all CompositeParticle objects (AllObjects) More...
class	CSCClusterRetriever
class	CscPrepDataRetriever
class	CTPDecisionRetriever
class	DataType
	Templated class to convert any object that is streamable in a ostringstream in a string. More...
struct	DispatchThreadArguments_t
class	ElectronRetriever
	Retrieves all Electron objects (ElectronAODCollection etc.) More...
class	EmTauROIRetriever
struct	Event_t
struct	EventRequest_t
class	EventStreamID
	For the client-server communication, each event is uniquely identified by the run number, the event number and a string determining which stream it belongs to. More...
class	ExternalONCRPCServerSvc
	This athena service will provide a link to an external ONCRPC server, e.g running as standalone application or in another athena application. More...
class	GeometryWriter
	Writes the inner detector and calorimeter geometry to an XML file for use with Atlantis. More...
class	IDataRetriever
	Interface for all DataRetriever classes, that extract event data from StoreGate and pass them on to a IFormatTool The interface will only hold the methods that have to be implemented by all DataRetrievers. More...
class	IEventReceiver
	Pure abstract interface for all event serving classes that provide events, streams and status information to the serving threads. More...
class	IEventServer
	Pure abstract interface for all event serving classes that provide events, streams and status information to the serving threads. More...
class	IFormatTool
	Abstract interface to event data formatting classes. More...
class	IGeometryWriter
class	IInDetGeoModelTool
	Interface for the InDetGeoModelTool. More...
class	IMessage
	Pure abstract interface to provide some athena-indepandant messaging Note that only MSG::Level enum is used from IMessageSvc. More...
class	InDetGeoModelTool
	Provides easy access to the geometry model managers and identifier helpers for all JiveXML::IDataRetriever implemenations in InDetJiveXML . More...
class	IServer
	Pure abstract interface for all full server implementations. More...
class	IServerSvc
	Athena interface for all ServerSvc implementations. More...
class	IStreamTool
	Abstract interface to JiveXML event streaming classes. More...
class	JetROIRetriever
	Retrieves ROI objects, default mode uses HLT/Offline object LVL1_ROI::jets_type. More...
class	JiveXMLServer
class	LArDigitRetriever
	Retrieves all Tile Calo Cell objects. More...
class	LVL1JetElementRetriever
class	LVL1ResultRetriever
class	LVL1TrigTowerRetriever
class	MdtPrepDataRetriever
class	MMPrepDataRetriever
class	MuonGeometryWriter
	Writes the GeoModel muon geometry to an XML file for use with Atlantis. More...
class	MuonRetriever
	Retrieves all Muon objects (StacoMuonCollection etc.) More...
class	MuonROIRetriever
	Retrieves ROI objects, default mode uses HLT/Offline object LVL1_ROI::emtaus_type. More...
class	ONCRPCServerSvc
	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...
class	PhotonRetriever
	Retrieves all Photon objects (PhotonAODCollection) More...
class	PixelClusterRetriever
	Retrieves all InDet::SiCluster objects. More...
class	PixelRDORetriever
	Retrieve all PixelRDORawData information (pixel raw hits) More...
class	RpcPrepDataRetriever
class	SCTRDORetriever
	Retrieve all SCT_RDORawData information (SCT raw hits) More...
class	SegmentRetriever
	Retrieves all Trk::SegmentCollection objects. More...
struct	ServerThreadArguments_t
class	SiClusterRetriever
	Retrieves all InDet::SiCluster objects. More...
class	SiSpacePointRetriever
	Retrieves all InDet::SpacePoint data for Pixel and SCT. More...
class	sTgcPrepDataRetriever
struct	Streams_t
class	StreamToFileTool
	Stream JiveXML event information into a file. More...
class	StreamToServerTool
	Stream event to an IServerSvc, which typically is a network server run from and controlled by the Athena application. More...
class	TauJetRetriever
	Retrieves all TauJet objects. More...
class	TgcPrepDataRetriever
class	ThreadCollection
	This class handles a collection of threads. More...
class	TrackParticleRetriever
	Retrieves all TrackParticle (AOD) Collection objects This is a chopped version of the full (RDO/ESD) Trk::Track retriever TrigJiveXML/TrackRetriever. More...
class	TrackRetriever
	Retrieves all Trk::TrackCollection objects. More...
class	TrigEMClusterRetriever
class	TriggerInfoRetriever
class	TrigMuonROIRetriever
class	TrigRpcDataRetriever
class	TrigSiSpacePointRetriever
class	TRTRetriever
	Retrieves all InDet::TRT_DriftCircle objects. More...
class	TruthMuonTrackRetriever
	Retrieves the muons information from the TrackRecordCollection. More...
class	TruthParticleRetriever
class	TruthTrackRetriever
	Retrieves the McEventCollection or the TrackRecordCollection in simulated cosmics. More...
class	VertexRetriever
	Retrieves all Trk::VxCandidate objects. More...
class	xAODCaloClusterRetriever
	Retrieves all Calo Cluster objects. More...
class	xAODElectronRetriever
	Retrieves all Electron objects (ElectronAODCollection etc.) More...
class	xAODEmTauROIRetriever
class	xAODJetRetriever
	Retrieves all Jet objects (JetAODCollection etc.) More...
class	xAODJetROIRetriever
class	xAODMissingETRetriever
	Retrieves all MissingET objects (MissingETAODCollection etc.) More...
class	xAODMuonRetriever
	Retrieves all Muon objects (MuonAODCollection etc.) More...
class	xAODMuonROIRetriever
class	xAODPhotonRetriever
	Retrieves all Photon objects (PhotonAODCollection etc.) More...
class	xAODTauRetriever
	Retrieves all Tau objects (TauAODCollection etc.) More...
class	xAODTrackParticleRetriever
	Retrieves all TrackParticle objects (TrackParticleAODCollection etc.) More...
class	xAODTrigDecisionRetriever
class	xAODTriggerTowerRetriever
	Retrieves TriggerTower xAOD objects. More...
class	xAODVertexRetriever
	Retrieves all Vertex objects (VertexAODCollection etc.) More...
class	XMLFormatTool
	Writes data stored in a DataMap into an XML stream in the format required by the Atlantis event display. More...

Typedefs
typedef std::vector< DataType >	DataVect
	Defines a map with a key and a vector of DataType objects e.g. More...
typedef std::map< std::string, DataVect >	DataMap
typedef std::pair< std::string, std::string >	TagType
	Defines a tag as a pair of strings. More...
typedef std::pair< const EventStreamID, const std::string >	EventStreamPair
	A map that stores events according to their EventStreamID Due to the way EventStreamID is build, events will be sorted by their stream name. More...
typedef std::map< const EventStreamID, const std::string >	EventStreamMap
typedef struct Streams_t	Streams
	Data structures for GetStreams functions. More...
typedef struct EventRequest_t	EventRequest
	Data structures for GetEvent functions. More...
typedef struct Event_t	Event
typedef struct ServerThreadArguments_t	ServerThreadArguments
	Arguments handed over fromt the main (Athena) thread to the server thread. More...
typedef struct DispatchThreadArguments_t	DispatchThreadArguments
	Arguments handed over fromt the main server thread to the thread dispatching the request (one for each request) More...

Functions
std::ostream &	operator<< (std::ostream &os, const DataType &dt)
	Allow this to be streamed in an ostream. More...
bool	checkResult (const int RetVal, const std::string &Module, IMessage *const ServerSvc)
	Simple result checking routine, that will output an errorMsg throught the ServerSvc if there was an error. More...
void	ReturnNull (SVCXPRT *transp, IServer *const ServerSvc)
	Implementation of NULLPROC Return nothing - server has just been pinged. More...
void	ReturnState (SVCXPRT *transp, IServer *const ServerSvc)
	Implementation of ONCRPC_ATHENASTATUS_PROC Return the current athena status in XDR representation. More...
void	ReturnStreams (SVCXPRT *transp, IServer *const ServerSvc)
	Implementation of ONCRPC_GETSTREAMS_PROC Return the currently available event streams. More...
void	ReturnEvent (SVCXPRT *transp, const EventRequest *eventReq, IServer *const ServerSvc)
	Implementation of ONCRPC_GETEVENT_PROC Return an event from a certain streams. More...
void	SetNewEvent (SVCXPRT *transp, const Event *event, IServer *const ServerSvc)
	Implementation of ONCRPC_SETEVENT_PROC Set a new event for a certain streams. More...
void *	ONCRPCServerThread (void *args)
	This is the actual server thread, which takes above arguments. More...
void *	ONCRPCDispatchThread (void *args)
	This is the thread handling the request - one thread per request. More...
void	ONCRPCRequestHandler (struct svc_req *rqstp, SVCXPRT *transp)
	The Request handler is called from the RPC kernel routines - hence extern C It creates a new dispatch thread and returns. More...
bool_t	xdr_streams (XDR *xdrsp, Streams *streams)
	De- and Encoding of Streams_t. More...
bool_t	xdr_event_req (XDR *xdrsp, EventRequest *eventReq)
	De-/Encoding of EventRequest_t. More...
bool_t	xdr_event (XDR *xdrsp, Event *event)
	De-/Encoding of Event_t. More...
void	ONCRPCCleanupHandler (void *args)
	This cleanup handler is called whenever the server thread exits. More...
void	manualPerigeeMatch (const Trk::Perigee *per1, const Trk::Perigee *per2, bool &matched)
	Manually match measured perigee d0,z0,phi0 of Trk::Track and Rec::TrackParticle, some problem with direct match depite identical values (maybe worth trying again). More...

Variables
const unsigned int	NSTREAMMAX = 1024
const unsigned int	NBYTESMAX = 0x1u << (sizeof( unsigned int )*8 - 1)
ThreadCollection *const	dispatchThreads = new ThreadCollection()
pthread_key_t ServerSvcKey	ATLAS_THREAD_SAFE

Detailed Description

This header is shared inbetween the C-style server thread and the C++ Athena ServerSvc.

Migrated to new TrigDecisionTool described here https://twiki.cern.ch/twiki/bin/view/Atlas/TrigDecisionTool15.

These are the XDR procedures that are used to encode/decode the data which is transfered inbetween client and server in the XDR (eXternal Data Representation).

It defines

• The constants by which RPC program, version and procedure are idenfied
• The definition of the data types exchanged inbetween the client and the server

format. This format is platform independant ( see standards RFC 1832 and RFC 4506).

Typedef Documentation

DataMap

typedef std::map<std::string, DataVect > JiveXML::DataMap

Definition at line 59 of file DataType.h.

DataVect

typedef std::vector<DataType> JiveXML::DataVect

Defines a map with a key and a vector of DataType objects e.g.

"CellID" and a vector of cell IDs

Definition at line 58 of file DataType.h.

DispatchThreadArguments

typedef struct DispatchThreadArguments_t JiveXML::DispatchThreadArguments

Arguments handed over fromt the main server thread to the thread dispatching the request (one for each request)

Definition at line 34 of file ONCRPCServerThreads.h.

Event

typedef struct Event_t JiveXML::Event

Definition at line 40 of file ONCRPCServer.h.

EventRequest

typedef struct EventRequest_t JiveXML::EventRequest

Data structures for GetEvent functions.

Definition at line 40 of file ONCRPCServer.h.

EventStreamMap

typedef std::map< const EventStreamID, const std::string> JiveXML::EventStreamMap

Definition at line 87 of file EventStream.h.

EventStreamPair

typedef std::pair< const EventStreamID, const std::string> JiveXML::EventStreamPair

A map that stores events according to their EventStreamID Due to the way EventStreamID is build, events will be sorted by their stream name.

std::map::find(key) and operator[] will return the event that belongs to this stream. In order to compare for identity of events use EventStreamID::isSameEvent()

Definition at line 86 of file EventStream.h.

ServerThreadArguments

typedef struct ServerThreadArguments_t JiveXML::ServerThreadArguments

Arguments handed over fromt the main (Athena) thread to the server thread.

Definition at line 17 of file ONCRPCServerThreads.h.

Streams

typedef struct Streams_t JiveXML::Streams

Data structures for GetStreams functions.

Definition at line 40 of file ONCRPCServer.h.

TagType

typedef std::pair<std::string,std::string> JiveXML::TagType

Defines a tag as a pair of strings.

Definition at line 62 of file DataType.h.

Function Documentation

checkResult()

bool JiveXML::checkResult	(	const int	RetVal,
		const std::string &	Module,
		IMessage *const	ServerSvc
	)

Simple result checking routine, that will output an errorMsg throught the ServerSvc if there was an error.

Returns false on error;

Definition at line 24 of file ONCRPCServerProcs.cxx.

                                                                                         {
 
    //Immediately return on no error
    if (RetVal == 0) return true ;
 
    //If there was an error, assemble the message
    std::ostringstream msg;
    char errbuf[256];
    msg << "Error in " << Module << ": " << strerror_r(RetVal, errbuf, sizeof(errbuf));
    
    //Output the message via the ServerSvc
    ServerSvc->Message(MSG::ERROR,msg.str());
 
    //In case of error return false
    return false ;
  }

manualPerigeeMatch()

void JiveXML::manualPerigeeMatch	(	const Trk::Perigee *	per1,
		const Trk::Perigee *	per2,
		bool &	matched
	)

Manually match measured perigee d0,z0,phi0 of Trk::Track and Rec::TrackParticle, some problem with direct match depite identical values (maybe worth trying again).

Definition at line 61 of file VertexRetriever.cxx.

                                                                {
 
    double diff1 = ( per1->parameters()[Trk::d0] -
                     per2->parameters()[Trk::d0] );
    double diff2 = ( per1->parameters()[Trk::z0] -
                     per2->parameters()[Trk::z0] );
    double diff3 = ( per1->parameters()[Trk::phi0] -
                     per2->parameters()[Trk::phi0] );
    double diff4 = ( per1->charge() -
                     per2->charge() );
    double diff5 = ( per1->pT() -
                     per2->pT() );
 
    matched = diff1+diff2+diff3+diff4+diff5 == 0.;
  }

ONCRPCCleanupHandler()

void JiveXML::ONCRPCCleanupHandler

(

void *

args

)

This cleanup handler is called whenever the server thread exits.

Definition at line 241 of file ONCRPCServerThreads.cxx.

                                       {
    //Since this cleanup handler is associated with the IServer* 
    //we will get exactly this as an argument
    ((JiveXML::IServer*)args)->Message(MSG::INFO,"Finished ONCRPC server thread");
    //Also call the servers callback
    ((JiveXML::IServer*)args)->ServerThreadStopped();
  }

ONCRPCDispatchThread()

void * JiveXML::ONCRPCDispatchThread

(

void *

args

)

This is the thread handling the request - one thread per request.

This is the dispatch function, which does the actual calls of the ServerSvc routines.

Many of these dispatch requests may be running at the same time, each in its own thread.

Definition at line 42 of file ONCRPCServerThreads.cxx.

                                           {
    
    //Get the thread arguments
    const struct svc_req* rqstp = ((DispatchThreadArguments*)args)->rqstp;
    const EventRequest* eventRequest = ((DispatchThreadArguments*)args)->evtReq;
    const Event* event = ((DispatchThreadArguments*)args)->evt;
    IServer* const ServerSvc = ((DispatchThreadArguments*)args)->ServerSvcPtr;
 
    //we have stored the arguments, so delete the object
    delete (DispatchThreadArguments*)args;
 
    try {
 
      //Now first of all register myself with the ServerSvc;
      dispatchThreads->AddThread(pthread_self());
      //Tell them the thread number if they want to know
      std::ostringstream msg; msg << "Created new dispatch thread " << pthread_self();
      ServerSvc->Message(MSG::VERBOSE,msg.str());
 
      //Process the request
      switch (rqstp->rq_proc){
        //NULLPROC - no data exchanged, just pinging for response
        case NULLPROC:
          ReturnNull(rqstp->rq_xprt, ServerSvc);
          break ; 
        //GETSTATUS - return the current athena status
        case ONCRPC_GETSTATUS_PROC:
          ReturnState(rqstp->rq_xprt, ServerSvc);
          break ;
        //GETSTREAMS - return the names of the available streams
        case ONCRPC_GETSTREAMS_PROC:
          ReturnStreams(rqstp->rq_xprt, ServerSvc);
          break ;
        //GETEVENT - return the current event in a given stream
        case ONCRPC_GETEVENT_PROC:
          ReturnEvent(rqstp->rq_xprt, eventRequest, ServerSvc);
          break ;
        //GETEVENT - return the current event in a given stream
        case ONCRPC_SETEVENT_PROC:
          SetNewEvent(rqstp->rq_xprt, event, ServerSvc);
          break ;
        default:
          std::ostringstream msg;
          msg << "Client request for server procedure #" << rqstp->rq_proc << " which is not defined";
          ServerSvc->Message(MSG::WARNING, msg.str());
          svcerr_noproc(rqstp->rq_xprt);
          break ;
      }
 
      //We have dealt with the request, delete it
      delete rqstp;
      delete event;
      delete eventRequest;
 
      //Finally unregister myself with the ServerSvc
      //Note that this will also detach the thread, so this has to be the last
      //statement before the thread returns
      dispatchThreads->RemoveThread(pthread_self());
 
    } catch (std::exception &e){
      std::ostringstream msg; msg << "Caught exception in DispatchThread: " << e.what();
      ServerSvc->Message(MSG::ERROR,msg.str());
    }
 
    return NULL;
  }

ONCRPCRequestHandler()

void JiveXML::ONCRPCRequestHandler	(	struct svc_req *	rqstp,
		SVCXPRT *
	)

The Request handler is called from the RPC kernel routines - hence extern C It creates a new dispatch thread and returns.

This function handles all the requests that are received from the server by creating a new thread for each request.

For request with an input value the input parameters are extracted first, as svc_getreqset is not thread safe.

Definition at line 116 of file ONCRPCServerThreads.cxx.

                                                                          {
 
    //Get the ServerSvc pointer for this thread
    JiveXML::IServer* const ServerSvc = (JiveXML::IServer*)pthread_getspecific(ServerSvcKey);
    std::ostringstream msg; msg << "Request handler in thread " << pthread_self();
    ServerSvc->Message(MSG::VERBOSE,msg.str()); msg.str("");
 
    try {
 
      //Check wether it is worth retrievin caller information
      if ( ServerSvc->LogLevel() <= MSG::DEBUG ){
 
        //Get information about the requester
        auto caller = (struct sockaddr* )svc_getcaller(rqstp->rq_xprt);
        char port[NI_MAXSERV]; 
        char host[NI_MAXHOST];
        char IPAddr[INET6_ADDRSTRLEN];
 
        //assemble a message
        msg << "Request from host ";
        //Add host name if we have one
        if(getnameinfo(caller, rqstp->rq_xprt->xp_addrlen,
                       host, sizeof host,
                       nullptr, 0,
                       0) == 0)
          msg << host << " ";
        //Add ip-address and port
        if(getnameinfo(caller, rqstp->rq_xprt->xp_addrlen,
                       IPAddr, sizeof IPAddr,
                       port, sizeof port,
                       NI_NUMERICSERV | NI_NUMERICHOST) == 0)
          msg << "(" << IPAddr << ") on port " << port;
        //Deliver the message
        ServerSvc->Message(MSG::DEBUG,msg.str()); msg.str("");
      }
 
      //Prepare the dispatch thread argument structure making a copy of the
      //request, so it sticks around if this thread is gone
      DispatchThreadArguments DpThreadArgs(ServerSvc, new svc_req(*rqstp), NULL, NULL);
 
      //Check for input data to the request
      switch (rqstp->rq_proc){
 
        case ONCRPC_GETEVENT_PROC: {
          //Create structure to hold request arguments on the head so we can
          //forward it to the dispatch thread.
          //NOTE: This is a C-struct that MUST be initalized.
          //It will be cleaned up by the dispatch thread
          EventRequest* evtReq = new EventRequest;
          evtReq->EventNumber=-2;evtReq->RunNumber=-2;evtReq->StreamName=NULL;
 
          //Retrieve the request arguments
          if (!svc_getargs(rqstp->rq_xprt,(xdrproc_t)xdr_event_req,(caddr_t)evtReq)){
              checkResult(errno,"creating dispatch tread arguments from GETEVENT call",ServerSvc);
              //tell the client his request failed
              svcerr_decode(rqstp->rq_xprt);
              //return immediately from this request
              return ;
           }
 
          //Set the event request argument
          DpThreadArgs.evtReq = evtReq;
          break;
        
        }  case ONCRPC_SETEVENT_PROC: {
          //Create structure to hold request arguments on the HEAP so we can
          //forward it to the dispatch thread.
          //NOTE: This is a C-struct that MUST be initalized
          //It will be cleaned up by the dispatch thread
          Event* evt = new Event;
          evt->EventNumber=-2;evt->RunNumber=-2;evt->StreamName=NULL;evt->NBytes=0;evt->EventData=NULL;
 
          //Retrieve the request arguments
          if (!svc_getargs(rqstp->rq_xprt,(xdrproc_t)xdr_event,(caddr_t)evt)){
              checkResult(errno,"creating dispatch tread arguments from SETEVENT call",ServerSvc);
              //tell the client his request failed
              svcerr_decode(rqstp->rq_xprt);
              //return immediately from this request
              return ;
           }
 
          //Set the event request argument
          DpThreadArgs.evt = evt;
          break;
        }
      }
        
      //Check for errors in all pthread functions
      int retVal = 0;
 
      //Generate thread attributes
      pthread_attr_t attr; retVal = pthread_attr_init (&attr);
      if ( ! checkResult(retVal,"request handler initializing thread attributes",ServerSvc)) return ;
 
      //Removing the limit on the thread memory usage as a test. Suspect that some threads do not have enough memory to finish and therefore eat up all the memory.
      //retVal = pthread_attr_setstacksize(&attr,10*PTHREAD_STACK_MIN);
      if ( ! checkResult(retVal,"request handler setting thread stacksize",ServerSvc)) return ;
      
      //NOTE: All threads are first created joinable, so we can wait for the to
      //finish using pthread_detach. Yet, when the thread removes itself from
      //the ThreadCollection, it will detach itself, so no memory is lost.
      retVal = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
      if ( ! checkResult(retVal,"request handler setting thread detach state",ServerSvc)) return;
 
      //Create a new thread
      pthread_t dispatchThread;
      retVal = pthread_create(&dispatchThread,&attr,&ONCRPCDispatchThread,new DispatchThreadArguments(DpThreadArgs));
      if ( ! checkResult(retVal,"request handler creating dispatch thread",ServerSvc)) return;
 
      //And wait till it has registered itself with the ThreadCollection
      dispatchThreads->WaitAdd();
 
      //release thread attributs
      retVal = pthread_attr_destroy(&attr);
      if ( ! checkResult(retVal,"request handler destroying thread attributes",ServerSvc)) return;
 
    } catch (std::exception &e){
      std::ostringstream msg; msg << "Caught exception in RequestHandler: " << e.what();
      ServerSvc->Message(MSG::ERROR,msg.str());
    }
  }

ONCRPCServerThread()

void * JiveXML::ONCRPCServerThread

(

void *

args

)

This is the actual server thread, which takes above arguments.

Main server therad.

Check wether the the portmapper is up and if it already has another instance of this server registered

Check wether the the portmapper is up and if it already has another instance of this server registered

Definition at line 252 of file ONCRPCServerThreads.cxx.

                                        {
 
    //Get a pointer to the Service that started this thread
    JiveXML::IServer* const ServerSvc  = (*( ServerThreadArguments* )args).ServerPtr;
    //Get the port number from the thread arguments
    const int PortNumber = (*( ServerThreadArguments* )args).PortNumber;
 
    //We've got the arguments and can delete the struct
    delete (ServerThreadArguments*)args;
    
    //Save the arguments in a thread-specific keys
    //Once the thread exits, the specified cleanup handler is called,
    //which in our case is used to notify the IServer of the thread termination
    int retVal = 0;
    retVal = pthread_key_create(&ServerSvcKey,ONCRPCCleanupHandler);
    if (! checkResult(retVal,"server thread creating thread-specific key",ServerSvc)) pthread_exit(NULL);
    retVal = pthread_setspecific(ServerSvcKey, ServerSvc);
    if (! checkResult(retVal,"server thread setting thread-specific key",ServerSvc)) pthread_exit(NULL);
 
    try {
 
      //Also install cleanup handlers if they are given as arguments
 
      //Send a message to the messaging function of the service
      ServerSvc->Message(MSG::INFO,"Started ONCRPC server thread");
 
      //Check wether the runServerThread flag is set
      if (!ServerSvc->GetRunServerFlag()){
        ServerSvc->Message(MSG::WARNING,"The run server flag is not set - stopping server thread immediately");
        pthread_exit(NULL);
      }
 
      //Get my address
      struct sockaddr_in my_addr; get_myaddress(&my_addr);
      //Try get the portnumber of that process
      unsigned short port = pmap_getport(&my_addr, ONCRPCSERVERPROG,ONCRPCSERVERVERS,IPPROTO_TCP);
      //Check if some program has already been registered
      if ( port != 0 ){
        //First report it
        std::ostringstream msg; msg << "Program with program number " << ONCRPCSERVERPROG 
                                   << " already registered with portmapper on local host";
        ServerSvc->Message(MSG::INFO,msg.str());
        //Next check if it is alive by creating a client and calling its NULLPROC
        CLIENT* client = clnt_create("localhost", ONCRPCSERVERPROG,ONCRPCSERVERVERS, "tcp");
        if (client != NULL){
// 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
          struct timeval timeout; timeout.tv_sec = 1; timeout.tv_usec = 0;
          clnt_stat ret = clnt_call(client, NULLPROC, (xdrproc_t)xdr_void, NULL, (xdrproc_t)xdr_void, NULL, timeout);
#if __GNUC__ >= 8
# pragma GCC diagnostic pop
#endif
          if (ret == RPC_SUCCESS){
            //So we already have a server, and it is alive -- then we don't start
            ServerSvc->Message(MSG::ERROR,"Server exists and is alive on local host - stopping this thread");
            pthread_exit(NULL);
          } else ServerSvc->Message(MSG::WARNING,"Existing server does not respond");
        } else ServerSvc->Message(MSG::WARNING,"Can not create client for existing server");
 
        //Now just force a clearing of the portmap entry
        ServerSvc->Message(MSG::WARNING,"Clearing existing portmap entry!");
        pmap_unset(ONCRPCSERVERPROG, ONCRPCSERVERVERS);
 
        //Finally clear the error entry by trying to get the port number again
        port = pmap_getport(&my_addr, ONCRPCSERVERPROG,ONCRPCSERVERVERS,IPPROTO_TCP);
      }
 
      //We should have left either the first or the second call to pmap_getport
      // with "program not registered" , which is returned as RPC_PROGNOTREGISTERD
      //or RPC_SYSTEMERROR. However, we could also have failed contacting the portmapper
      if ((rpc_createerr.cf_stat != RPC_PROGNOTREGISTERED) &&
          (rpc_createerr.cf_stat != RPC_SYSTEMERROR )){
        ServerSvc->Message(MSG::ERROR,clnt_spcreateerror("Failed querying portmapper on local host for existing servers"));
        pthread_exit(NULL);
      } 
 
      //The socket to connect to, not bound to a specifc port by default
      int server_socket = RPC_ANYSOCK;
 
      //Check if a specific port was requested
      if (PortNumber != 0){
 
        //Create a socket
        if ((server_socket = socket(AF_INET, SOCK_STREAM, 0)) != 0) {
          checkResult(errno,"server thread creating socket",ServerSvc);
          pthread_exit(NULL);
        }
 
        //Create an inet address with the given port number
        struct sockaddr_in server_addr;    
        server_addr.sin_family = AF_INET;          //IPv4 address
        server_addr.sin_addr.s_addr = INADDR_ANY;  //will bind all interfaces
        server_addr.sin_port = htons(PortNumber);  //port number in network byte order   
        memset(&server_addr.sin_zero, 0, sizeof(server_addr.sin_zero)); //zero padding
 
        //Now bind the socket to that inet address
        if (bind(server_socket, (struct sockaddr *)&server_addr, sizeof(struct sockaddr)) != 0){
          std::ostringstream msg; msg << "server thread binding socket to port " << PortNumber;
          checkResult(errno,msg.str(),ServerSvc);
          pthread_exit(NULL);
        }
        std::ostringstream msg; msg << "Successfully bound port " << PortNumber;
        ServerSvc->Message(MSG::INFO,msg.str());
      }
      
      //Get a transport handle using default buffer sizes
      SVCXPRT * transp = svctcp_create(server_socket, 0, 0);
 
      //Check result
      if (transp == NULL) {
        ServerSvc->Message(MSG::ERROR,"Opening TCP socket failed");
        pthread_exit(NULL);
      }
 
      //Next register our server with the RPC daemon
#ifndef __APPLE__
      if (! svc_register(transp,ONCRPCSERVERPROG,ONCRPCSERVERVERS,ONCRPCRequestHandler,IPPROTO_TCP)){
#else
      if (! svc_register(transp,ONCRPCSERVERPROG,ONCRPCSERVERVERS,(void (*)())ONCRPCRequestHandler,IPPROTO_TCP)){
#endif
        ServerSvc->Message(MSG::ERROR,"Could not register ONCRPC Server with RPC daemon");
        checkResult(errno,"registering ONCRPC Server with RPC daemon",ServerSvc);
        pthread_exit(NULL);
      }
     
      //A set of socket file descriptors
      fd_set SocketFileDescriptorSet;
      //Size of the socket handle table
      int NFileDescriptors = FD_SETSIZE;
 
      //Count the number of requests we have received while running
      unsigned long NRequests = 0;
 
      do {  
 
        ServerSvc->Message(MSG::VERBOSE,"Waiting for call...");
 
        //Get the file descriptor set
        SocketFileDescriptorSet = svc_fdset;
 
        //Wait for input on any of these using select:
        // - first NULL: not cecking for readiness to write to socket
        // - second NULL: not checking for socket conditions
        // - third NULL: timeout - wait forever
        int ret = select(NFileDescriptors,&SocketFileDescriptorSet,NULL,NULL,NULL);
        // Check result
        switch (ret){
          // an error occured
          case -1: 
            // If just an interrupt continue
            if ( errno == EINTR ) continue ;
            // Else show error
            checkResult(errno,"server thread returning from select",ServerSvc);
            return NULL;
          // a timeout occured - break the loop
          case 0:
            break ;
          // activity on one of the sockets
          default:
            //Count number of requests
            ++NRequests;
            // Get the requests from the sockets
            svc_getreqset(&SocketFileDescriptorSet);
        }
      } while ( ServerSvc->GetRunServerFlag() );
 
      ServerSvc->Message(MSG::DEBUG,"Serving loop finished");
 
      //Unregister the service, so no clients can connect anymore
      svc_unregister(ONCRPCSERVERPROG,ONCRPCSERVERVERS);
      
      //Also stop all the dispatching threads, which need to finish before this
      //one. Otherwise the server will become invalid, before all requests are
      //served
      
      //Tell how many threads we are waiting for
      std::ostringstream msg; msg << "Waiting for " << dispatchThreads->NumberOfThreads() << " open dispatch threads to terminate ... ";
      ServerSvc->Message(MSG::INFO,msg.str());
      //Then wait for them all to finish
      dispatchThreads->JoinAll();
      //Tell when we are done
      ServerSvc->Message(MSG::INFO, " ... finished all dispatch threads");
 
      //Now that the threads are stopped we can finish the transport protocol and
      //close the port
      svc_destroy(transp);
 
      //Return number of requests - as all local variables will disappear, 
      //the return value needs to be allocated on the heap;
      unsigned long* NReqPtr = new unsigned long(NRequests);
      //Terminate this thread
      pthread_exit(NReqPtr);
 
    } catch (std::exception &e){
      std::ostringstream msg; msg << "Caught exception in ServerThread: " << e.what();
      ServerSvc->Message(MSG::ERROR,msg.str());
    }
 
    return NULL;
  }

operator<<()

std::ostream & JiveXML::operator<<	(	std::ostream &	os,
		const DataType &	dt
	)

Allow this to be streamed in an ostream.

Allow the DataType to be streamed in an ostream.

Parameters

os	the stream
dt	the DataType object

Definition at line 27 of file DataType.cxx.

                                                              {
    return os << dt.toString();
  }

ReturnEvent()

void JiveXML::ReturnEvent	(	SVCXPRT *	transp,
		const EventRequest *	eventReq,
		IServer *const	ServerSvc
	)

Implementation of ONCRPC_GETEVENT_PROC Return an event from a certain streams.

Return the actual event for a requested stream, only if.

• the stream is available
• this is not the same event as the user had already asked for
• otherwise, set flags accordingly in the return structure

Definition at line 105 of file ONCRPCServerProcs.cxx.

                                                                                             {
 
    //Check if there was an event request
    std::ostringstream msg;
    if (! eventReq ) {
      msg << "Request to return event with void event request structure";
      ServerSvc->Message(MSG::ERROR,msg.str());
      svcerr_systemerr(transp);
      return ;
    }
   
    //Be a bit verbose
    msg << "Request for stream " << eventReq->StreamName
        << " with existing event " << eventReq->EventNumber 
        << " from run " << eventReq->RunNumber;
    //Send message an clear stream
    ServerSvc->Message(MSG::DEBUG,msg.str()); msg.str("");
 
    //Next get the current eventStreamID from that stream
    EventStreamID currEvtStreamID = ServerSvc->GetEventStreamID(eventReq->StreamName);
 
    //Also get an event stream ID for the requested event
    EventStreamID reqEvtStreamID(eventReq->EventNumber,eventReq->RunNumber,eventReq->StreamName);
 
    //We got all the data from the request, so delete it
    caddr_t nc_eventReq ATLAS_THREAD_SAFE = (caddr_t)eventReq;
    xdr_free((xdrproc_t)xdr_event_req,nc_eventReq);
 
    //Prepare response structure
    Event event;
    event.isAvailable = false ;
    event.isIdentical = false ;
    event.isCompressed = false ;
    event.EventNumber = currEvtStreamID.EventNumber();
    event.RunNumber = currEvtStreamID.RunNumber();
    event.StreamName = currEvtStreamID.StreamNameCStr();
    event.NBytes = 0; event.EventData = NULL ;
 
    //Check if such a stream was currently found
    if (!currEvtStreamID.isValid()) {
      // Sending reply with "no such stream" - i.e unmodified event structure
      msg << "Sending response NO SUCH STREAM "; ServerSvc->Message(MSG::DEBUG,msg.str()); 
      if (!svc_sendreply(transp, (xdrproc_t)xdr_event, (caddr_t)&event))
        checkResult(errno,"dispatch thread sending reply to GETEVENT call",ServerSvc);
      return ;
    } else
      //We have an event in that stream
      event.isAvailable = true ;
 
    //Check if this is the same event the user already has
    if ( currEvtStreamID.isSameEvent(reqEvtStreamID) ){
      // Set the same event flag
      event.isIdentical = true ;
      // Sending reply with "same event"
      msg << "Sending response SAME EVENT "; ServerSvc->Message(MSG::DEBUG,msg.str()); 
      if (!svc_sendreply(transp, (xdrproc_t)xdr_event, (caddr_t)&event))
        checkResult(errno,"dispatch thread sending reply to GETEVENT call",ServerSvc);
      return ;
    } 
 
    //Otherwise just get the event itself
    std::string evt = ServerSvc->GetEvent( currEvtStreamID );
    event.EventData = evt.c_str(); 
    event.NBytes = strlen(evt.c_str())+1; //including null-termination character
    //Send reply with an actual event
    msg << "Sending response NEW EVENT (" << event.EventNumber << ","<< event.RunNumber << ")" ;
    ServerSvc->Message(MSG::DEBUG,msg.str()); 
    if (!svc_sendreply(transp, (xdrproc_t)xdr_event, (caddr_t)&event))
      checkResult(errno,"dispatch thread sending reply to GETEVENT call",ServerSvc);
 
    //No need to use xdr_free to free the returned event object since we did
    //not allocate resources for it (just point the pointers to the proper place).
  }

ReturnNull()

void JiveXML::ReturnNull	(	SVCXPRT *	transp,
		IServer *const	ServerSvc
	)

Implementation of NULLPROC Return nothing - server has just been pinged.

Definition at line 45 of file ONCRPCServerProcs.cxx.

                                                              {
// 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 (!svc_sendreply(transp, (xdrproc_t)xdr_void, 0))
      //check global errno variable on result
      checkResult(errno,"dispatch thread sending reply to NULLPROC call",ServerSvc);
  }

ReturnState()

void JiveXML::ReturnState	(	SVCXPRT *	transp,
		IServer *const	ServerSvc
	)

Implementation of ONCRPC_ATHENASTATUS_PROC Return the current athena status in XDR representation.

Implementation of ONCRPC_GETSTATUS_PROC Return the current athena status in XDR representation.

Definition at line 63 of file ONCRPCServerProcs.cxx.

                                                               {
    int state = ServerSvc->GetState();
    if (!svc_sendreply(transp, (xdrproc_t)xdr_int, (caddr_t)&state))
      //check global errno variable on result
      checkResult(errno,"dispatch thread sending reply to GETSTATUS call",ServerSvc);
  }

ReturnStreams()

void JiveXML::ReturnStreams	(	SVCXPRT *	transp,
		IServer *const	ServerSvc
	)

Implementation of ONCRPC_GETSTREAMS_PROC Return the currently available event streams.

Return the names of all the streams that are available at the server.

Definition at line 73 of file ONCRPCServerProcs.cxx.

                                                                 {
 
    // First of all get the list of names
    std::vector<std::string> StreamNames = ServerSvc->GetStreamNames();
 
    // Create a new XDR data structure to fill in the streams
    Streams XDRStreams;
 
    // Fill in number of streams
    XDRStreams.NStreams = StreamNames.size();
 
    //Create a new array with the stream names
    XDRStreams.StreamNames = (char**) malloc(StreamNames.size()*sizeof(char*));
 
    //Fill in all the names themselves
    for ( unsigned int iName = 0; iName < StreamNames.size(); ++iName)
      XDRStreams.StreamNames[iName] = strdup(StreamNames[iName].c_str());
 
    //return that object
    if (!svc_sendreply(transp, (xdrproc_t)xdr_streams, (caddr_t)&XDRStreams))
      checkResult(errno,"dispatch thread sending reply to GETSTREAMS call",ServerSvc);
 
    //Afterwards free the object
    xdr_free((xdrproc_t)xdr_streams,(caddr_t)&XDRStreams);
  }

SetNewEvent()

void JiveXML::SetNewEvent	(	SVCXPRT *	transp,
		const Event *	event,
		IServer *const	ServerSvc
	)

Implementation of ONCRPC_SETEVENT_PROC Set a new event for a certain streams.

Set a new event for a given stream.

Definition at line 182 of file ONCRPCServerProcs.cxx.

                                                                                   {
 
    //Check if there was a valid event request
    //NOTE: Event number can be 0, run number can not
    std::ostringstream msg;
    if ( (!event) || (event->EventNumber<0) || (!event->RunNumber) || (!event->StreamName) )  {
      msg << "Request to set new event with void or invalid event structure";
      ServerSvc->Message(MSG::ERROR,msg.str());
      svcerr_systemerr(transp);
      return ;
    }
 
    //Be a bit verbose
    msg << "Request to set new event  " << event->EventNumber 
        << " from run " << event->RunNumber
        << " for stream " << event->StreamName;
    //Send message an clear stream
    ServerSvc->Message(MSG::DEBUG,msg.str()); msg.str("");
 
    //The success flag returned to the client 
    //Need to use bool_t (4bytes) instead of bool (1byte)!
    bool_t isSuccess=false;
 
    //Create a new EventStreamID from the data we have got
    EventStreamID newEvtStreamID(event->EventNumber,event->RunNumber,event->StreamName);
 
    //Make sure this ID is valid
    if (! newEvtStreamID.isValid()) { 
      msg << "Request to set new event with invalid stream name or event/run number";
      ServerSvc->Message(MSG::ERROR,msg.str());
      if (!svc_sendreply(transp, (xdrproc_t)xdr_bool, (caddr_t)&isSuccess))
        checkResult(errno,"dispatch thread sending reply to SETEVENT call",ServerSvc);
      return ;
    }
 
    //And make sure there is data
    if ((! event->EventData ) || (strlen(event->EventData) == 0)){
      msg << "Request to set new event with invalid event data";
      ServerSvc->Message(MSG::ERROR,msg.str());
      if (!svc_sendreply(transp, (xdrproc_t)xdr_bool, (caddr_t)&isSuccess))
        checkResult(errno,"dispatch thread sending reply to SETEVENT call",ServerSvc);
      return ;
    }
 
    //Also make sure it has the proper length
    if ( strlen(event->EventData) != event->NBytes -1 ){
      msg << "Event string length does not match claimed number of bytes (null-termination problem?)";
      ServerSvc->Message(MSG::ERROR,msg.str());
      if (!svc_sendreply(transp, (xdrproc_t)xdr_bool, (caddr_t)&isSuccess))
        checkResult(errno,"dispatch thread sending reply to SETEVENT call",ServerSvc);
      return ;
    }
    //Get the event data
    std::string EvtData(event->EventData);
   
    //Now set the event at the server
    if (ServerSvc->UpdateEventForStream( newEvtStreamID, EvtData).isSuccess())
      //and set the success flag
      isSuccess = true;
 
    //Show the result
    msg << "Sending response for set new event: ";
    if (isSuccess) msg << " SUCCESS!"; else msg << " FAILED!";
    ServerSvc->Message(MSG::DEBUG,msg.str()); 
    //Send the reply
    if (!svc_sendreply(transp, (xdrproc_t)xdr_bool, (caddr_t)&isSuccess))
      checkResult(errno,"dispatch thread sending reply to SETEVENT call",ServerSvc);
 
    //Afterwards free the object
    caddr_t nc_event ATLAS_THREAD_SAFE = (caddr_t)event;
    xdr_free((xdrproc_t)xdr_event,nc_event);
  }

xdr_event()

bool_t JiveXML::xdr_event	(	XDR *	xdrsp,
		Event *	event
	)

De-/Encoding of Event_t.

Definition at line 51 of file ONCRPCXDRProcs.cxx.

                                              {
 
    //de-/encode is-available bit
    if (! xdr_bool( xdrsp, &(event->isAvailable)))
      return false ;
 
    //de-/encode is-available bit
    if (! xdr_bool( xdrsp, &(event->isIdentical)))
      return false ;
 
    //de-/encode is-available bit
    if (! xdr_bool( xdrsp, &(event->isCompressed)))
      return false ;
    
    //de-/encode stream name
    if (! xdr_wrapstring( xdrsp, (char**)&(event->StreamName)))
      return false ;
 
    //de-/encode event number
    if (! xdr_int64_t( xdrsp, &(event->EventNumber)))
      return false ;
    
    //de-/encode run number
    if (! xdr_int64_t( xdrsp, &(event->RunNumber)))
      return false ;
 
    // wrap data as an array of bytes with length NBytes - this allows non-null
    // terminated byte arrays to be send. Also - in contrast to an array - no
    // subroutine is called for each byte. Instead, the string is just padded
    // with null-char to obtain a length divisible by four
   return (xdr_bytes( xdrsp, (caddr_t*)&event->EventData, &event->NBytes, NBYTESMAX));
  }

xdr_event_req()

bool_t JiveXML::xdr_event_req	(	XDR *	xdrsp,
		EventRequest *	eventReq
	)

De-/Encoding of EventRequest_t.

Implementation of XDR converters for ONCRPC_GETEVENT_PROC Return the currently available eventRequest or event in XDR representation.

Definition at line 32 of file ONCRPCXDRProcs.cxx.

                                                            {
 
    //de-/encode event number
    if (! xdr_int64_t( xdrsp, &(eventReq->EventNumber)))
      return false ;
    
    //de-/encode run number
    if (! xdr_int64_t( xdrsp, &(eventReq->RunNumber)))
      return false ;
 
    //de-/encode stream name
    if (! xdr_wrapstring( xdrsp, (char**)&(eventReq->StreamName)))
      return false ;
 
    // All done successfully
    return true ;
  }

xdr_streams()

bool_t JiveXML::xdr_streams	(	XDR *	xdrsp,
		Streams *	streams
	)

De- and Encoding of Streams_t.

Implementation of XDR converters for ONCRPC_GETSTREAMS_PROC Return the currently available event streams in XDR representation.

Definition at line 19 of file ONCRPCXDRProcs.cxx.

                                                    {
 
    // As we do not know the string length, we use xdr_wrap_string
    return (xdr_array( xdrsp, (caddr_t*)&streams->StreamNames, &streams->NStreams,
            NSTREAMMAX, sizeof(char*), (xdrproc_t)xdr_wrapstring));
  }

Variable Documentation

ATLAS_THREAD_SAFE

pthread_key_t ServerSvcKey JiveXML::ATLAS_THREAD_SAFE

Definition at line 35 of file ONCRPCServerThreads.cxx.

dispatchThreads

ThreadCollection *const JiveXML::dispatchThreads = new ThreadCollection()

Definition at line 31 of file ONCRPCServerThreads.cxx.

NBYTESMAX

const unsigned int JiveXML::NBYTESMAX = 0x1u << (sizeof( unsigned int )*8 - 1)

Definition at line 40 of file ONCRPCServer.h.

NSTREAMMAX

const unsigned int JiveXML::NSTREAMMAX = 1024

Definition at line 38 of file ONCRPCServer.h.