Loading...
Searching...
No Matches
XrdOfsTPCProg Class Reference

`#include <XrdOfsTPCProg.hh>`

Collaboration diagram for XrdOfsTPCProg:

## Public Member Functions

| XrdOfsTPCProg (XrdOfsTPCProg *Prev, int num, int errMon)
| ~XrdOfsTPCProg ()
void | Cancel ()
void | Run ()
int | Xeq (bool &isIPv4)

## Static Public Member Functions

static int | Init ()
static XrdOfsTPCProg * | Start (XrdOfsTPCJob *jP, int &rc)

## Detailed Description

Definition at line 40 of file XrdOfsTPCProg.hh.

## Constructor & Destructor Documentation

## XrdOfsTPCProg()

XrdOfsTPCProg::XrdOfsTPCProg | ( | XrdOfsTPCProg * | Prev,
|  | int | num,
|  | int | errMon )

Definition at line 122 of file XrdOfsTPCProg.cc.

: Prog(&OfsEroute, errMon),
JobStream(&OfsEroute),
Next(Prev), Job(0)
{snprintf(Pname, sizeof(Pname), "TPC job %d: ", num);
Pname[sizeof(Pname)-1] = 0;
}
OfsEroute
XrdSysError OfsEroute(0)

References XrdOfsTPCProg(), and OfsEroute.

Referenced by XrdOfsTPCProg(), Init(), and Start().

Here is the call graph for this function:
Here is the caller graph for this function:

## ~XrdOfsTPCProg()

XrdOfsTPCProg::~XrdOfsTPCProg | ( |  | ) |

inline

Definition at line 57 of file XrdOfsTPCProg.hh.

{}

## Member Function Documentation

## Cancel()

void XrdOfsTPCProg::Cancel | ( |  | ) |

inline

Definition at line 44 of file XrdOfsTPCProg.hh.

{JobStream.Drain();}

## Init()

int XrdOfsTPCProg::Init | ( |  | ) |

static

Definition at line 167 of file XrdOfsTPCProg.cc.

{
int n;
// Allocate copy program objects
//
for (n = 0; n < Cfg.xfrMax; n++)
{pgmIdle = new XrdOfsTPCProg(pgmIdle, n, Cfg.errMon);
if (pgmIdle->Prog.Setup(Cfg.XfrProg, &OfsEroute)) return 0;
}
// All done
//
Cfg.doEcho = Cfg.doEcho || GTRACE(debug);
return 1;
}
GTRACE
#define GTRACE(act)
Definition XrdBwmTrace.hh:40
XrdOfsTPCProg::XrdOfsTPCProg
XrdOfsTPCProg(XrdOfsTPCProg *Prev, int num, int errMon)
Definition XrdOfsTPCProg.cc:122
XrdOfsTPCParms::Cfg
XrdOfsTPCConfig Cfg
Definition XrdOfsTPC.cc:85
XrdOfsTPCConfig::xfrMax
int xfrMax
Definition XrdOfsTPCConfig.hh:47
XrdOfsTPCConfig::errMon
int errMon
Definition XrdOfsTPCConfig.hh:48
XrdOfsTPCConfig::XfrProg
char * XfrProg
Definition XrdOfsTPCConfig.hh:39
XrdOfsTPCConfig::doEcho
bool doEcho
Definition XrdOfsTPCConfig.hh:50

References XrdOfsTPCProg(), XrdOfsTPCParms::Cfg, GTRACE, and OfsEroute.

Referenced by XrdOfsTPC::Start().

Here is the call graph for this function:
Here is the caller graph for this function:

## Run()

void XrdOfsTPCProg::Run | ( |  | ) |

Definition at line 188 of file XrdOfsTPCProg.cc.

{
XrdXrootdTpcMon::TpcInfo monInfo;
struct stat Stat;
const char *clID, *at;
char *questSrc, *questLfn, *questDst;
int rc;
bool isIPv4, doMon = Cfg.tpcMon != 0;
char clBuff[592];
// Run the current job and indicate it's ending status and possibly getting a
// another job to run. Note "Job" will always be valid.
//
do{if (doMon)
{monInfo.Init();
gettimeofday(&monInfo.begT, 0);
}
rc = Xeq(isIPv4);
if (doMon)
{gettimeofday(&monInfo.endT, 0);
if ((questSrc = index(Job->Info.Key, '?'))) *questSrc = 0;
monInfo.srcURL = Job->Info.Key;
if ((questLfn = index(Job->Info.Lfn, '?'))) *questLfn = 0;
monInfo.dstURL = Job->Info.Lfn;
monInfo.endRC = rc;
if (Job->Info.Str) monInfo.strm = Job->Info.Str;
if (isIPv4) monInfo.opts |= XrdXrootdTpcMon::TpcInfo::isIPv4;
clID = Job->Info.Org;
if (clID && (at = index(clID, '@')) && !index(at+1, '.'))
{const char *dName = XrdNetIdentity::Domain();
if (dName)
{snprintf(clBuff, sizeof(clBuff), "%s%s", clID, dName);
clID = clBuff;
}
}
monInfo.clID = clID;
if ((questDst = index(Job->Info.Dst, '?'))) *questDst = 0;
if (!XrdOfsOss->Stat(Job->Info.Dst, &Stat)) monInfo.fSize = Stat.st_size;
if (questDst) *questDst = '?';
Cfg.tpcMon->Report(monInfo);
if (questLfn) *questLfn = '?';
if (questSrc) *questSrc = '?';
}
Job = Job->Done(this, eRec, rc);
} while(Job);
// No more jobs to run. Place us on the idle queue. Upon return this thread
// will end.
//
pgmMutex.Lock();
Next = pgmIdle;
pgmIdle = this;
pgmMutex.UnLock();
}
Stat
struct stat Stat
Definition XrdCks.cc:49
XrdOfsOss
XrdOss * XrdOfsOss
Definition XrdOfs.cc:163
stat
#define stat(a, b)
Definition XrdPosix.hh:101
XrdNetIdentity::Domain
static const char * Domain(const char **eText=0)
Definition XrdNetIdentity.cc:177
XrdOfsTPCProg::Xeq
int Xeq(bool &isIPv4)
Definition XrdOfsTPCProg.cc:280
XrdOss::Stat
virtual int Stat(const char *path, struct stat *buff, int opts=0, XrdOucEnv *envP=0)=0
XrdXrootdTpcMon::Report
void Report(TpcInfo &info)
Definition XrdXrootdTpcMon.cc:124
XrdOfsTPCConfig::tpcMon
XrdXrootdTpcMon * tpcMon
Definition XrdOfsTPCConfig.hh:37
XrdXrootdTpcMon::TpcInfo::fSize
size_t fSize
Definition XrdXrootdTpcMon.hh:51
XrdXrootdTpcMon::TpcInfo::clID
const char * clID
Definition XrdXrootdTpcMon.hh:46
XrdXrootdTpcMon::TpcInfo::isIPv4
static const int isIPv4
Definition XrdXrootdTpcMon.hh:58
XrdXrootdTpcMon::TpcInfo::endT
struct timeval endT
Definition XrdXrootdTpcMon.hh:48
XrdXrootdTpcMon::TpcInfo::srcURL
const char * srcURL
Definition XrdXrootdTpcMon.hh:49
XrdXrootdTpcMon::TpcInfo::endRC
int endRC
Definition XrdXrootdTpcMon.hh:52
XrdXrootdTpcMon::TpcInfo::begT
struct timeval begT
Definition XrdXrootdTpcMon.hh:47
XrdXrootdTpcMon::TpcInfo::dstURL
const char * dstURL
Definition XrdXrootdTpcMon.hh:50
XrdXrootdTpcMon::TpcInfo::strm
unsigned char strm
Definition XrdXrootdTpcMon.hh:54
XrdXrootdTpcMon::TpcInfo::Init
void Init()
Definition XrdXrootdTpcMon.hh:60
XrdXrootdTpcMon::TpcInfo::opts
unsigned short opts
Definition XrdXrootdTpcMon.hh:53

References XrdXrootdTpcMon::TpcInfo::begT, XrdOfsTPCParms::Cfg, XrdXrootdTpcMon::TpcInfo::clID, XrdNetIdentity::Domain(), XrdXrootdTpcMon::TpcInfo::dstURL, XrdXrootdTpcMon::TpcInfo::endRC, XrdXrootdTpcMon::TpcInfo::endT, XrdXrootdTpcMon::TpcInfo::fSize, XrdXrootdTpcMon::TpcInfo::Init(), XrdXrootdTpcMon::TpcInfo::isIPv4, XrdXrootdTpcMon::TpcInfo::opts, XrdXrootdTpcMon::TpcInfo::srcURL, Stat, stat, XrdXrootdTpcMon::TpcInfo::strm, Xeq(), and XrdOfsOss.

Referenced by XrdOfsTPCProgRun().

Here is the call graph for this function:
Here is the caller graph for this function:

## Start()

XrdOfsTPCProg * XrdOfsTPCProg::Start | ( | XrdOfsTPCJob * | jP,
|  | int & | rc )

static

Definition at line 253 of file XrdOfsTPCProg.cc.

{
XrdSysMutexHelper pgmMon(&pgmMutex);
XrdOfsTPCProg *pgmP;
pthread_t tid;
// Get a new program object, if none left, tell the caller to try later
//
if (!(pgmP = pgmIdle)) {rc = 0; return 0;}
pgmP->Job = jP;
// Start a thread to run the job
//
if ((rc = XrdSysThread::Run(&tid, XrdOfsTPCProgRun, (void *)pgmP, 0,
"TPC job")))
return 0;
// We are all set, return the program being used
//
pgmIdle = pgmP->Next;
return pgmP;
}
XrdOfsTPCProgRun
void * XrdOfsTPCProgRun(void *pp)
Definition XrdOfsTPCProg.cc:77
XrdSysThread::Run
static int Run(pthread_t *, void *(*proc)(void *), void *arg, int opts=0, const char *desc=0)
Definition XrdSysPthread.cc:307

References XrdOfsTPCProg(), XrdSysThread::Run(), and XrdOfsTPCProgRun().

Referenced by XrdOfsTPCJob::Sync().

Here is the call graph for this function:
Here is the caller graph for this function:

## Xeq()

int XrdOfsTPCProg::Xeq | ( | bool & | isIPv4 | ) |

Definition at line 280 of file XrdOfsTPCProg.cc.

{
EPNAME("Xeq");
credFile cFile(Job);
const char *Args[6], *eVec[6], **envArg;
char *lP, *Colon, *cksVal, sBuff[8], *tident = Job->Info.Org;
char *Quest = index(Job->Info.Key, '?');
int i, rc, aNum = 0;
// If we have credentials, write them out to a file
//
if (cFile.Path && (rc = ExportCreds(cFile.Path)))
{strcpy(eRec, "Copy failed; unable to pass credentials.");
return rc;
}
// Echo out what we are doing if so desired
//
if (Cfg.doEcho)
{if (Quest) *Quest = 0;
OfsEroute.Say(Pname,tident," copying ",Job->Info.Key," to ",Job->Info.Dst);
if (Quest) *Quest = '?';
}
// Determine checksum option
//
cksVal = (Job->Info.Cks ? Job->Info.Cks : Cfg.cksType);
if (cksVal)
{Args[aNum++] = "-C";
Args[aNum++] = cksVal;
}
// Set streams option if need be
//
if (Job->Info.Str)
{sprintf(sBuff, "%d", static_cast<int>(Job->Info.Str));
Args[aNum++] = "-S";
Args[aNum++] = sBuff;
}
// Set remaining arguments
//
Args[aNum++] = Job->Info.Key;
Args[aNum++] = Job->Info.Dst;
// Always export the trace identifier of the original issuer
//
char tidBuff[512];
snprintf(tidBuff, sizeof(tidBuff), "XRD_TIDENT=%s", tident);
eVec[0] = tidBuff;
envArg = eVec;
i = 1;
// Export source protocol if present
//
char sprBuff[128];
if (Job->Info.Spr)
{snprintf(sprBuff, sizeof(sprBuff), "XRDTPC_SPROT=%s", Job->Info.Spr);
eVec[i++] = sprBuff;
}
// Export target protocol if present
//
char tprBuff[128];
if (Job->Info.Tpr)
{snprintf(tprBuff, sizeof(tprBuff), "XRDTPC_TPROT=%s", Job->Info.Tpr);
eVec[i++] = tprBuff;
}
// If we need to reproxy, export the path
//
char rpxBuff[1024];
if (Job->Info.Rpx)
{snprintf(rpxBuff, sizeof(rpxBuff), "XRD_CPTARGET=%s", Job->Info.Rpx);
eVec[i++] = rpxBuff;
}
// Determine if credentials are being passed, If so, pass where it is.
//
if (cFile.Path) eVec[i++] = cFile.pEnv;
eVec[i] = 0;
// Start the job.
//
if ((rc = Prog.Run(&JobStream, Args, aNum, envArg)))
{strcpy(eRec, "Copy failed; unable to start job.");
OfsEroute.Emsg("TPC", Job->Info.Org, Job->Info.Lfn, eRec);
return rc;
}
// Now we drain the output looking for an end of run line. This line should
// be printed as an error message should the copy fail.
//
*eRec = 0;
isIPv4 = false;
while((lP = JobStream.GetLine()))
{if (!strcmp(lP, "!-!IPv4")) isIPv4 = true;
if ((Colon = index(lP, ':')) && *(Colon+1) == ' ')
{strncpy(eRec, Colon+2, sizeof(eRec)-1);
eRec[sizeof(eRec)-1] = 0;
}
if (Cfg.doEcho && *lP) OfsEroute.Say(Pname, lP);
}
// The job has completed. So, we must get the ending status.
//
if ((rc = Prog.RunDone(JobStream)) < 0) rc = -rc;
DEBUG(Pname <<"ended with rc=" <<rc);
// Check if we should generate a message
//
if (rc && !(*eRec)) sprintf(eRec, "Copy failed with return code %d", rc);
// Log failures and optionally remove the file (Info would do that as well
// but much later on, so we do it now).
//
if (rc)
{OfsEroute.Emsg("TPC", Job->Info.Org, Job->Info.Lfn, eRec);
if (Cfg.autoRM) XrdOfsOss->Unlink(Job->Info.Lfn);
} else Job->Info.Success();
// All done
//
return rc;
}
tident
#define tident
Definition XrdBwmHandle.cc:109
DEBUG
#define DEBUG(x)
Definition XrdBwmTrace.hh:54
EPNAME
#define EPNAME(x)
Definition XrdBwmTrace.hh:56
XrdOss::Unlink
virtual int Unlink(const char *path, int Opts=0, XrdOucEnv *envP=0)=0
XrdSysError::Emsg
int Emsg(const char *esfx, int ecode, const char *text1, const char *text2=0)
Definition XrdSysError.cc:95
XrdSysError::Say
void Say(const char *text1, const char *text2=0, const char *txt3=0, const char *text4=0, const char *text5=0, const char *txt6=0)
Definition XrdSysError.cc:141
XrdOfsTPCConfig::cksType
char * cksType
Definition XrdOfsTPCConfig.hh:40
XrdOfsTPCConfig::autoRM
bool autoRM
Definition XrdOfsTPCConfig.hh:51

References XrdOfsTPCParms::Cfg, DEBUG, EPNAME, OfsEroute, tident, and XrdOfsOss.

Referenced by Run().

Here is the caller graph for this function:

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

- XrdOfsTPCProg.hh
- XrdOfsTPCProg.cc