Loading...
Searching...
No Matches
XrdOfsTPCJob.cc
Go to the documentation of this file.
/******************************************************************************/
/* */
/* X r d O f s T P C J o b . c c */
/* */
/* (c) 2012 by the Board of Trustees of the Leland Stanford, Jr., University */
/* All Rights Reserved */
/* Produced by Andrew Hanushevsky for Stanford University under contract */
/* DE-AC02-76-SFO0515 with the Department of Energy */
/* */
/* This file is part of the XRootD software suite. */
/* */
/* XRootD is free software: you can redistribute it and/or modify it under */
/* the terms of the GNU Lesser General Public License as published by the */
/* Free Software Foundation, either version 3 of the License, or (at your */
/* option) any later version. */
/* */
/* XRootD is distributed in the hope that it will be useful, but WITHOUT */
/* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or */
/* FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public */
/* License for more details. */
/* */
/* You should have received a copy of the GNU Lesser General Public License */
/* along with XRootD in a file called COPYING.LESSER (LGPL license) and file */
/* COPYING (GPL license). If not, see <http://www.gnu.org/licenses/>. */
/* */
/* The copyright holder's institutional names and contributor's names may not */
/* be used to endorse or promote products derived from this software without */
/* specific prior written permission of the institution or contributor. */
/******************************************************************************/
#include "XrdOfs/XrdOfsStats.hh"
#include "XrdOfs/XrdOfsTPCJob.hh"
#include "XrdOfs/XrdOfsTPCProg.hh"
#include "XrdOuc/XrdOucCallBack.hh"
#include "XrdSfs/XrdSfsInterface.hh"
/******************************************************************************/
/* G l o b a l O b j e c t s */
/******************************************************************************/
extern XrdSysError OfsEroute;
extern XrdOfsStats OfsStats;
/******************************************************************************/
/* S t a t i c O b j e c t s */
/******************************************************************************/
XrdSysMutex XrdOfsTPCJob::jobMutex;
XrdOfsTPCJob *XrdOfsTPCJob::jobQ = 0;
XrdOfsTPCJob *XrdOfsTPCJob::jobLast = 0;
/******************************************************************************/
/* C o n s t r u c t o r */
/******************************************************************************/
XrdOfsTPCJob::XrdOfsTPCJob(const char *Url, const char *Org,
const char *Lfn, const char *Pfn,
const char *Cks, short lfnLoc[2],
const char *Spr, const char *Tpr)
: XrdOfsTPC(Url, Org, Lfn, Pfn, Cks, Spr, Tpr),
Next(0), myProg(0), eCode(0), Status(isWaiting)
{ lfnPos[0] = lfnLoc[0]; lfnPos[1] = lfnLoc[1]; }
/******************************************************************************/
/* D e l */
/******************************************************************************/
void XrdOfsTPCJob::Del()
{
XrdOfsTPCJob *pP = 0;
bool tpcCan = false;
// Remove from queue if we are still in the queue
//
jobMutex.Lock();
if (inQ)
{if (this == jobQ) jobQ = Next;
else {pP = jobQ;
while(pP && pP->Next != this) pP = pP->Next;
if (pP) pP->Next = Next;
}
if (this == jobLast) jobLast = pP;
inQ = 0; tpcCan = true;
} else if (Status == isRunning && myProg)
{myProg->Cancel(); tpcCan = true;}
if (tpcCan && Info.cbP)
{Refs++; // Make sure this object cannot get deleted
Info.Reply(SFS_ERROR, ECANCELED, "destination file prematurely closed",
&jobMutex); // Mutex is unlocked upon return!
jobMutex.Lock();
Refs--; // Undo the extra ref increase
}
// Delete the element if possible
//
if (Refs <= 1) {jobMutex.UnLock(); delete this;}
else {Refs--; jobMutex.UnLock();}
}
/******************************************************************************/
/* D o n e */
/******************************************************************************/
XrdOfsTPCJob *XrdOfsTPCJob::Done(XrdOfsTPCProg *pgmP, const char *eTxt, int rc)
{
XrdSysMutexHelper jobMon(&jobMutex);
XrdOfsTPCJob *jP;
// Indicate job status
//
eCode = rc; Status = isDone;
if (Info.Key) free(Info.Key);
Info.Key = (rc ? strdup(eTxt) : 0);
// Check if we need to do a callback
//
if (Info.cbP)
{if (rc) Info.Reply(SFS_ERROR, rc, eTxt);
else Info.Reply(SFS_OK, 0, "");
}
// Check if anyone is waiting for a program
//
if ((jP = jobQ))
{if (jP == jobLast) jobQ = jobLast = 0;
else jobQ = jP->Next;
jP->myProg = pgmP; jP->Refs++; jP->inQ = 0; jP->Status = isRunning;
if (jP->Info.cbP) jP->Info.Reply(SFS_OK, 0, "");
}
// Free up this job and return the next job, if any
//
myProg = 0;
if (Refs <= 1) delete this;
else Refs--;
return jP;
}
/******************************************************************************/
/* S y n c */
/******************************************************************************/
int XrdOfsTPCJob::Sync(XrdOucErrInfo *eRR)
{
static const int cbWaitTime = 1800;
XrdSysMutexHelper jobMon(&jobMutex);
int rc;
// If we are running then simply wait for the copy to complete
//
if (Status == isRunning)
{if (Info.SetCB(eRR)) return SFS_ERROR;
eRR->setErrCode(cbWaitTime);
Info.Engage();
return SFS_STARTED;
}
// If we are done then return what we have (this can't change)
//
if (Status == isDone)
{if (eCode) {eRR->setErrInfo(eCode, Info.Key); return SFS_ERROR;}
return SFS_OK;
}
// The only thing left is that we are an unstarted job, so try to start it.
//
if (inQ) {myProg = 0; rc = 0;}
else if ((myProg = XrdOfsTPCProg::Start(this, rc)))
{Refs++; Status = isRunning; return SFS_OK;}
// We could not allocate a program to this job. Check if this is due to an err
//
if (rc)
{OfsEroute.Emsg("TPC", rc, "create tpc job thread");
Status = isDone;
eCode = ECANCELED;
if (Info.Key) free(Info.Key);
Info.Key = strdup("Copy failed; resources unavailable.");
return Info.Fail(eRR, "resources unavailable", ECANCELED);
}
// No programs available, place this job in callback mode
//
if (Info.SetCB(eRR)) return SFS_ERROR;
if (jobLast) {jobLast->Next = this; jobLast = this;}
else jobQ = jobLast = this;
inQ = 1;
eRR->setErrCode(cbWaitTime);
Info.Engage();
return SFS_STARTED;
}
XrdOfsStats.hh
XrdOfsTPCJob.hh
XrdOfsTPCProg.hh
OfsEroute
XrdSysError OfsEroute(0)
OfsStats
XrdOfsStats OfsStats
Definition XrdOfs.cc:113
XrdOucCallBack.hh
XrdSfsInterface.hh
SFS_ERROR
#define SFS_ERROR
Definition XrdSfsInterface.hh:109
SFS_STARTED
#define SFS_STARTED
Definition XrdSfsInterface.hh:111
SFS_OK
#define SFS_OK
Definition XrdSfsInterface.hh:108
XrdOfsStats
Definition XrdOfsStats.hh:37
XrdOfsTPCInfo::cbP
XrdOucCallBack * cbP
Definition XrdOfsTPCInfo.hh:95
XrdOfsTPCInfo::Reply
void Reply(int rC, int eC, const char *eMsg, XrdSysMutex *mP=0)
Definition XrdOfsTPCInfo.cc:133
XrdOfsTPCJob
Definition XrdOfsTPCJob.hh:39
XrdOfsTPCJob::Del
void Del()
Definition XrdOfsTPCJob.cc:68
XrdOfsTPCJob::XrdOfsTPCJob
XrdOfsTPCJob(const char *Url, const char *Org, const char *Lfn, const char *Pfn, const char *Cks, short lfnLoc[2], const char *Spr, const char *Tpr)
Definition XrdOfsTPCJob.cc:56
XrdOfsTPCJob::Done
XrdOfsTPCJob * Done(XrdOfsTPCProg *pgmP, const char *eTxt, int rc)
Definition XrdOfsTPCJob.cc:105
XrdOfsTPCJob::Sync
int Sync(XrdOucErrInfo *eRR)
Definition XrdOfsTPCJob.cc:144
XrdOfsTPCProg
Definition XrdOfsTPCProg.hh:41
XrdOfsTPCProg::Start
static XrdOfsTPCProg * Start(XrdOfsTPCJob *jP, int &rc)
Definition XrdOfsTPCProg.cc:253
XrdOfsTPC::XrdOfsTPC
XrdOfsTPC()
Definition XrdOfsTPC.hh:99
XrdOfsTPC::Refs
char Refs
Definition XrdOfsTPC.hh:129
XrdOfsTPC::Info
XrdOfsTPCInfo Info
Definition XrdOfsTPC.hh:109
XrdOfsTPC::inQ
char inQ
Definition XrdOfsTPC.hh:130
XrdOucErrInfo
Definition XrdOucErrInfo.hh:101
XrdOucErrInfo::setErrInfo
int setErrInfo(int code, const char *emsg)
Definition XrdOucErrInfo.hh:147
XrdOucErrInfo::setErrCode
int setErrCode(int code)
Definition XrdOucErrInfo.hh:136
XrdSysError
Definition XrdSysError.hh:90
XrdSysMutexHelper
Definition XrdSysPthread.hh:263
XrdSysMutex
Definition XrdSysPthread.hh:165