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;
             }

Referenced by Init().

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();}

References XrdOucStream::Drain().

Referenced by XrdOfsTPCJob::Del().

Here is the call graph for this function:

Here is the caller graph for this function:

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;
}

References XrdOfsTPCProg(), XrdNetPMarkConfig::Cfg, GTRACE, OfsEroute, and XrdOucProg::Setup().

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();
}

References XrdXrootdTpcMon::TpcInfo::begT, XrdNetPMarkConfig::Cfg, XrdXrootdTpcMon::TpcInfo::clID, XrdNetIdentity::Domain(), XrdOfsTPCJob::Done(), XrdOfsTPCInfo::Dst, XrdXrootdTpcMon::TpcInfo::dstURL, XrdXrootdTpcMon::TpcInfo::endRC, XrdXrootdTpcMon::TpcInfo::endT, XrdXrootdTpcMon::TpcInfo::fSize, XrdOfsTPC::Info, XrdXrootdTpcMon::TpcInfo::Init(), XrdXrootdTpcMon::TpcInfo::isIPv4, XrdOfsTPCInfo::Key, XrdOfsTPCInfo::Lfn, XrdSysMutex::Lock(), XrdXrootdTpcMon::TpcInfo::opts, XrdOfsTPCInfo::Org, XrdXrootdTpcMon::TpcInfo::srcURL, Stat, stat, XrdOss::Stat(), XrdOfsTPCInfo::Str, XrdXrootdTpcMon::TpcInfo::strm, XrdSysMutex::UnLock(), 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;
}

References 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;
}

References XrdNetPMarkConfig::Cfg, XrdOfsTPCInfo::Cks, DEBUG, XrdOfsTPCInfo::Dst, XrdSysError::Emsg(), EPNAME, XrdOucStream::GetLine(), XrdOfsTPC::Info, XrdOfsTPCInfo::Key, XrdOfsTPCInfo::Lfn, OfsEroute, XrdOfsTPCInfo::Org, XrdOfsTPCInfo::Rpx, XrdOucProg::Run(), XrdOucProg::RunDone(), XrdSysError::Say(), XrdOfsTPCInfo::Spr, XrdOfsTPCInfo::Str, XrdOfsTPCInfo::Success(), tident, XrdOfsTPCInfo::Tpr, XrdOss::Unlink(), and XrdOfsOss.

Referenced by Run().

Here is the call graph for this function:

Here is the caller graph for this function:

---

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

• XrdOfsTPCProg.hh
• XrdOfsTPCProg.cc