#include <XrdXrootdGSReal.hh>

Inheritance diagram for XrdXrootdGSReal:

Collaboration diagram for XrdXrootdGSReal:

Classes
struct	GSParms

Public Member Functions
	XrdXrootdGSReal (const GSParms &gsParms, bool &aOK)

	~XrdXrootdGSReal ()
	Destructor. Normally, this object is never deleted. More...

void	DoIt ()

void	Flush ()

uint32_t	GetDictID (const char *text, bool isPath=false)

bool	HasHdr ()

void	Ident ()

bool	Insert (const char *data, int dlen)

bool	Insert (int dlen)

char *	Reserve (int dlen)

int	SetAutoFlush (int afsec)

int	Space ()

Public Member Functions inherited from XrdJob
	XrdJob (const char *desc="")

virtual	~XrdJob ()

Public Member Functions inherited from XrdXrootdGStream
	XrdXrootdGStream (XrdXrootdGSReal &gsRef)

void	Flush ()

uint32_t	GetDictID (const char *text, bool isPath=false)

bool	HasHdr ()

bool	Insert (const char *data, int dlen)

bool	Insert (int dlen)

char *	Reserve (int dlen)

int	SetAutoFlush (int afsec)

int	Space ()

Public Member Functions inherited from XrdXrootdMonitor::Hello
	Hello (const char *dest, char mode)

virtual	~Hello ()

Static Public Attributes
static const int	fmtBin = 1
	Do not include info. More...

static const int	fmtCgi = 2
	Format as binary info. More...

static const int	fmtJson = 3
	Format as CGI info. More...

static const int	fmtNone = 0

static const int	hdrFull = 5
	Include site, host, port, inst, pgm. More...

static const int	hdrHost = 3
	Include site, host. More...

static const int	hdrInst = 4
	Include site, host, port, inst. More...

static const int	hdrNone = 0
	Format as JSON info. More...

static const int	hdrNorm = 1
	Include standard header. More...

static const int	hdrSite = 2
	Include site. More...

static const int	optNoID = 0x01
	Don't send ident records. More...

Static Public Attributes inherited from XrdXrootdGStream
static const int	MaxDataLen = 65280
	The larest amount of data that can be inserted in a single call to GStream. More...

Additional Inherited Members
Static Public Member Functions inherited from XrdXrootdMonitor::Hello
static bool	Hail ()

Public Attributes inherited from XrdJob
const char *	Comment

XrdJob *	NextJob

Protected Member Functions inherited from XrdXrootdGStream
	~XrdXrootdGStream ()
	Destructor. This stream should never be directly deleted. More...

Detailed Description

Definition at line 48 of file XrdXrootdGSReal.hh.

Class Documentation

◆ XrdXrootdGSReal::GSParms

struct XrdXrootdGSReal::GSParms

Definition at line 95 of file XrdXrootdGSReal.hh.

Collaboration diagram for XrdXrootdGSReal::GSParms:

Class Members
const char *	dest	Destination for records.
int	flsT	Flush time (default from monitor)
char	Fmt	How to handle the records.
char	Hdr	Hdr type.
int	maxL	Maximum packet length (default 32K)
int	Mode	the monitor type for send routing.
char	Opt	Options.
const char *	pin	the plugin name.
kXR_char	Type	the specific G-Stream identifier

Constructor & Destructor Documentation

◆ XrdXrootdGSReal()

XrdXrootdGSReal::XrdXrootdGSReal	(	const GSParms &	gsParms,
		bool &	aOK
	)

Definition at line 64 of file XrdXrootdGSReal.cc.

                                 : XrdJob("GStream"), XrdXrootdGStream(*this),
                                   Hello(gsParms.Opt & XrdXrootdGSReal::optNoID
                                     || gsParms.Hdr == XrdXrootdGSReal::hdrNone
                                      ? 0 : gsParms.dest, gsParms.Fmt),
                                   pSeq(0), pSeqID(0), pSeqDID(0), binHdr(0),
                                   isCGI(false)
 {
    static const int minSZ = 1024;
    static const int dflSZ = 1024*32;
    static const int maxSZ = 1024*64;
    int flsT, maxL, hdrLen;
  
 // Do common initialization
 //
    memset(&hInfo, 0, sizeof(hInfo));
    aOK = true;
  
 // Compute the correct size of the UDP buffer
 //
    if (gsParms.maxL <= 0) maxL = dflSZ;
       else if (gsParms.maxL < minSZ) maxL = minSZ;
               else if (gsParms.maxL > maxSZ) maxL = maxSZ;
                       else maxL = gsParms.maxL;
    maxL &= ~7; // Doubleword lengths
  
 // Allocate the UDP buffer. Try to keep the data within a single page.
 //
    int align;
    if (maxL >= getpagesize())         align = getpagesize();
       else if (maxL >= 2048)          align = 2048;
               else if (maxL >= 1024)  align = 1024;
                       else            align = sizeof(void*);
  
    if (posix_memalign((void **)&udpBuffer, align, maxL)) {aOK = false; return;}
  
 // Setup the header as needed
 //
    if (gsParms.Hdr == hdrNone)
       {hdrLen = 0;
        binHdr = 0;
        dictHdr = idntHdr0 = idntHdr1 = 0;
       } else {
        switch(gsParms.Fmt)
              {case fmtBin:  hdrLen = hdrBIN(gsParms);
                             break;
               case fmtCgi:  hdrLen = hdrCGI(gsParms, udpBuffer, maxL);
                             break;
               case fmtJson: hdrLen = hdrJSN(gsParms, udpBuffer, maxL);
                             break;
               default:      hdrLen = 0;
              }
        if (gsParms.Opt & optNoID)
           {if (idntHdr0) {free(idntHdr0); idntHdr0 = 0;}
            if (idntHdr1) {free(idntHdr1); idntHdr1 = 0;}
           }
       }
  
 // Setup buffer pointers
 //
    udpBFirst = udpBNext = udpBuffer + hdrLen;
    udpBEnd = udpBuffer + maxL - 1;
  
    tBeg = tEnd = afTime = 0;
  
 // Initialize remaining variables
 //
    monType = gsParms.Mode;
    rsvbytes = 0;
  
 // If we have a specific end-point, then create a network relay to it
 //
    if (gsParms.dest) udpDest = new XrdNetMsg(eDest, gsParms.dest, &aOK, true);
       else udpDest = 0;
  
 // Setup autoflush (a negative value uses the default)
 //
    if (gsParms.flsT < 0) flsT = XrdXrootdMonitor::Flushing();
       else flsT = gsParms.flsT;
    afRunning = false;
    SetAutoFlush(flsT);
  
 // Construct our user name as in <gNamePI>.0:0@<myhost>
 //
    char idBuff[1024];
    snprintf(idBuff, sizeof(idBuff), "%s.0:0@%s", gsParms.pin, monHost);
  
 // Register ourselves
 //
    gMon.Register(idBuff, monHost, "xroot");
 }

References XrdXrootdGSReal::GSParms::dest, XrdXrootdMonInfo::eDest, XrdXrootdGSReal::GSParms::flsT, XrdXrootdMonitor::Flushing(), XrdXrootdGSReal::GSParms::Fmt, fmtBin, fmtCgi, fmtJson, XrdXrootdGSReal::GSParms::Hdr, hdrNone, XrdXrootdGSReal::GSParms::maxL, XrdXrootdGSReal::GSParms::Mode, XrdXrootdMonInfo::monHost, XrdXrootdGSReal::GSParms::Opt, optNoID, XrdXrootdGSReal::GSParms::pin, XrdXrootdMonitor::User::Register(), and SetAutoFlush().

Here is the call graph for this function:

◆ ~XrdXrootdGSReal()

XrdXrootdGSReal::~XrdXrootdGSReal ( )

inline

Destructor. Normally, this object is never deleted.

Definition at line 112 of file XrdXrootdGSReal.hh.

112 {}

Member Function Documentation

◆ DoIt()

void XrdXrootdGSReal::DoIt ( )

virtual

Implements XrdJob.

Definition at line 173 of file XrdXrootdGSReal.cc.

 {
    XrdSysMutexHelper gHelp(gMutex);
  
 // Check if we need to do anything here
 //
    afRunning = false;
    if (afTime)
       {if (tBeg && time(0)-tBeg >= afTime) Expel(0);
        AutoFlush();
       }
 }

◆ Flush()

void XrdXrootdGSReal::Flush ( )

Definition at line 241 of file XrdXrootdGSReal.cc.

 {
    XrdSysMutexHelper gHelp(gMutex);
    Expel(0);
 }

Referenced by XrdXrootdGStream::Flush().

Here is the caller graph for this function:

◆ GetDictID()

uint32_t XrdXrootdGSReal::GetDictID	(	const char *	text,
		bool	isPath = `false`
	)

Definition at line 251 of file XrdXrootdGSReal.cc.

 {
 // If this is binary encoded, the record the mapping and return it
 //
    if (binHdr) return (isPath ? gMon.MapPath(text) : gMon.MapInfo(text));
  
 // If there are no headers then we can't produce this record
 //
    uint32_t psq, did = XrdXrootdMonitor::GetDictID(true);
    if (!dictHdr) return htonl(did);
  
 // We need to do some additional work to generate non-binary headers here
 //
    struct iovec iov[3];
    char dit = (isPath ? XROOTD_MON_MAPPATH : XROOTD_MON_MAPINFO);
    char buff[1024];
  
 // Generate a new packet sequence number
 //
    gMutex.Lock();
    if (pSeqDID >= 999) pSeqDID = 0;
       else pSeqDID++;
    psq = pSeqDID;
    gMutex.UnLock();
  
 // Generate the packet
 //
    iov[0].iov_base = buff;
    iov[0].iov_len  = snprintf(buff, sizeof(buff), dictHdr, dit, psq, did);
    iov[1].iov_base = (void *)text;
    iov[1].iov_len  = strlen(text);
    iov[2].iov_base = (void *)"\"}";
    iov[2].iov_len  = 3;
  
 // Now send it off
 //
    udpDest->Send(iov, (*dictHdr == '{' ? 3 : 2));
    return htonl(did);
 }

References XrdXrootdMonitor::GetDictID(), XrdSysMutex::Lock(), XrdXrootdMonitor::User::MapInfo(), XrdXrootdMonitor::User::MapPath(), XrdNetMsg::Send(), XrdSysMutex::UnLock(), XROOTD_MON_MAPINFO, and XROOTD_MON_MAPPATH.

Referenced by XrdXrootdGStream::GetDictID().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ HasHdr()

bool XrdXrootdGSReal::HasHdr ( )

Definition at line 295 of file XrdXrootdGSReal.cc.

 {
    return binHdr != 0 || dictHdr != 0;
 }

Referenced by XrdXrootdGStream::HasHdr().

Here is the caller graph for this function:

◆ Ident()

void XrdXrootdGSReal::Ident ( )

virtual

Reimplemented from XrdXrootdMonitor::Hello.

Definition at line 455 of file XrdXrootdGSReal.cc.

 {
    struct iovec iov[2];
    char buff[40];
    uint32_t psq;
  
 // If identification suppressed, then just return
 //
    if (!idntHdr0 || !udpDest) return;
  
 // Generate a new packet sequence number
 //
    gMutex.Lock();
    if (pSeqID >= 999) pSeqID = 0;
       else pSeqID++;
    psq = pSeqID;
    gMutex.UnLock();
  
 // Create header and iovec to send the header
 //
    iov[0].iov_base = buff;
    iov[0].iov_len  = snprintf(buff, sizeof(buff), idntHdr0, psq);
    iov[1].iov_base = idntHdr1;
    iov[1].iov_len  = idntHsz1;
    udpDest->Send(iov, 2);
 }

References XrdSysMutex::Lock(), XrdNetMsg::Send(), and XrdSysMutex::UnLock().

Here is the call graph for this function:

◆ Insert() [1/2]

bool XrdXrootdGSReal::Insert	(	const char *	data,
		int	dlen
	)

Definition at line 486 of file XrdXrootdGSReal.cc.

 {
  
 // Validate the length and message
 //
    if (dlen < 8 || dlen > XrdXrootdGStream::MaxDataLen
    ||  !data    || data[dlen-1]) return false;
  
 // Reserve the storage and copy the message. It always will end with a newline
 //
    gMutex.Lock();
    Expel(dlen);
    memcpy(udpBNext, data, dlen-1);
    udpBNext[dlen-1] = '\n';
  
 // Timestamp the record and aAdjust buffer pointers
 //
    tEnd = time(0);
    if (udpBNext == udpBFirst) tBeg = tEnd;
    udpBNext += dlen;
  
 // All done
 //
    gMutex.UnLock();
    return true;
 }

References XrdSysMutex::Lock(), XrdXrootdGStream::MaxDataLen, and XrdSysMutex::UnLock().

Referenced by XrdXrootdGStream::Insert().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ Insert() [2/2]

bool XrdXrootdGSReal::Insert ( int dlen )

Definition at line 515 of file XrdXrootdGSReal.cc.

 {
    XrdSysMutexHelper gHelp(gMutex);
  
 // Make sure space is reserved
 //
    if (!rsvbytes) return false;
  
 // We are now sure that the recursive lock is held twice by this thread. So,
 // make it a unitary lock so it gets fully unlocked upon rturn.
 //
    gMutex.UnLock();
  
 // Check for cancellation
 //
    if (!dlen)
       {rsvbytes = 0;
        return true;
       }
  
 // Length, it must >= 8 and <= reserved amount and the data must end with a 0.
 //
    if (dlen > rsvbytes || dlen < 8 || *(udpBNext+dlen-1))
       {rsvbytes = 0;
        return false;
       }
  
 // Adjust the buffer space and time stamp the record
 //
    tEnd = time(0);
    if (udpBNext == udpBFirst) tBeg = tEnd;
    udpBNext += dlen;
    *(udpBNext-1) = '\n';
    rsvbytes = 0;
  
 // All done
  
    return true;
 }

References XrdSysMutex::UnLock().

Here is the call graph for this function:

◆ Reserve()

char * XrdXrootdGSReal::Reserve ( int dlen )

Definition at line 559 of file XrdXrootdGSReal.cc.

 {
 // Validate the length
 //
    if (dlen < 8 || dlen > XrdXrootdGStream::MaxDataLen) return 0;
  
 // Make sure there is no reserve outstanding
 //
    gMutex.Lock();
    if (rsvbytes)
       {gMutex.UnLock();
        return 0;
       }
  
 // Return the allocated the space but keep the lock until Insert() is called.
 //
    rsvbytes = dlen;
    Expel(dlen);
    return udpBNext;
 }

References XrdSysMutex::Lock(), XrdXrootdGStream::MaxDataLen, and XrdSysMutex::UnLock().

Referenced by XrdXrootdGStream::Reserve().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ SetAutoFlush()

int XrdXrootdGSReal::SetAutoFlush ( int afsec )

Definition at line 584 of file XrdXrootdGSReal.cc.

 {
    XrdSysMutexHelper gHelp(gMutex);
  
 // Save the current settting and establish the new one and relaunch
 //
    int afNow = afTime;
    afTime = (afsec > 0 ? afsec : 0);
    AutoFlush();
  
 // All done
 //
    return afNow;
 }

Referenced by XrdXrootdGSReal(), and XrdXrootdGStream::SetAutoFlush().

Here is the caller graph for this function:

◆ Space()

int XrdXrootdGSReal::Space ( )

Definition at line 603 of file XrdXrootdGSReal.cc.

 {
    XrdSysMutexHelper gHelp(gMutex);
   
 // Return amount of space left
 //
    return udpBEnd - udpBNext;
 }

Referenced by XrdXrootdGStream::Space().

Here is the caller graph for this function:

Member Data Documentation

◆ fmtBin

const int XrdXrootdGSReal::fmtBin = 1

static

Do not include info.

Definition at line 82 of file XrdXrootdGSReal.hh.

Referenced by XrdXrootdGSReal().

◆ fmtCgi

const int XrdXrootdGSReal::fmtCgi = 2

static

Format as binary info.

Definition at line 83 of file XrdXrootdGSReal.hh.

Referenced by XrdXrootdGSReal().

◆ fmtJson

const int XrdXrootdGSReal::fmtJson = 3

static

Format as CGI info.

Definition at line 84 of file XrdXrootdGSReal.hh.

Referenced by XrdXrootdGSReal().

◆ fmtNone

const int XrdXrootdGSReal::fmtNone = 0

static

Constructor

Parameters

gsParms	the stream parameters as defined by GSParms.
aOK	reference to a boolean which will contain true on success or will be set to false, otherwise.

Definition at line 81 of file XrdXrootdGSReal.hh.

◆ hdrFull

const int XrdXrootdGSReal::hdrFull = 5

static

Include site, host, port, inst, pgm.

Definition at line 91 of file XrdXrootdGSReal.hh.

◆ hdrHost

const int XrdXrootdGSReal::hdrHost = 3

static

Include site, host.

Definition at line 89 of file XrdXrootdGSReal.hh.

◆ hdrInst

const int XrdXrootdGSReal::hdrInst = 4

static

Include site, host, port, inst.

Definition at line 90 of file XrdXrootdGSReal.hh.

◆ hdrNone

const int XrdXrootdGSReal::hdrNone = 0

static

Format as JSON info.

Do not include header

Definition at line 86 of file XrdXrootdGSReal.hh.

Referenced by XrdXrootdGSReal().

◆ hdrNorm

const int XrdXrootdGSReal::hdrNorm = 1

static

Include standard header.

Definition at line 87 of file XrdXrootdGSReal.hh.

◆ hdrSite

const int XrdXrootdGSReal::hdrSite = 2

static

Include site.

Definition at line 88 of file XrdXrootdGSReal.hh.

◆ optNoID

const int XrdXrootdGSReal::optNoID = 0x01

static

Don't send ident records.

Definition at line 93 of file XrdXrootdGSReal.hh.

Referenced by XrdXrootdGSReal().

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

Classes

Public Member Functions

Static Public Attributes

Additional Inherited Members

Detailed Description

Class Documentation

◆ XrdXrootdGSReal::GSParms

Constructor & Destructor Documentation

◆ XrdXrootdGSReal()

◆ ~XrdXrootdGSReal()

Member Function Documentation

◆ DoIt()

◆ Flush()

◆ GetDictID()

◆ HasHdr()

◆ Ident()

◆ Insert() [1/2]

◆ Insert() [2/2]

◆ Reserve()

◆ SetAutoFlush()

◆ Space()

Member Data Documentation

◆ fmtBin

◆ fmtCgi

◆ fmtJson

◆ fmtNone

◆ hdrFull

◆ hdrHost

◆ hdrInst

◆ hdrNone

◆ hdrNorm

◆ hdrSite

◆ optNoID