32 #include <arpa/inet.h>
33 #include <sys/types.h>
56 #define DEBUGXQ(x) DEBUG(rID<<sessN<<rspstID[urState]<<reqstID[myState]<<x)
58 #define DUMPIT(x,y) XrdSsiUtils::b2x(x,y,hexBuff,sizeof(hexBuff),dotBuff)<<dotBuff
81 {
" [new",
" [begun",
" [bound",
86 {
" wtReq] ",
" xqReq] ",
" wtRsp] ",
87 " doRsp] ",
" odRsp] ",
" erRsp] "
97 int XrdSsiFileReq::freeCnt = 0;
98 int XrdSsiFileReq::freeMax = 256;
110 DEBUGXQ((oP ?
"oucbuff" :
"sfsbuff") <<
" rqsz=" <<rSz);
144 DEBUGXQ(msgLen <<
" byte alert presented wtr=" <<respWait);
156 if (msgLen <= 0 || haveResp || isEnding)
172 {alrtLast->next = aP;
175 alrtPend = alrtPend->
next;
179 if (alrtLast) alrtLast->next = aP;
207 freeReq = nP->nextReq;
223 snprintf(nP->rID,
sizeof(nP->rID),
"%u:", rnum);
237 void XrdSsiFileReq::BindDone()
243 DEBUGXQ(
"Bind called; for request " <<reqID);
254 {
case isBegun: urState = isBound;
255 case isBound:
return;
257 case isDone:
if (!schedDone)
269 Log.
Emsg(epname,
tident,
"Invalid req/rsp state; giving up on object!");
276 void XrdSsiFileReq::Dispose()
282 DEBUGXQ(
"Recycling request...");
307 {
case isNew: myState = xqReq; urState = isBegun;
308 DEBUGXQ(
"Calling service processor");
315 case isAbort:
DEBUGXQ(
"Skipped calling service processor");
321 case isDone: cancel = (myState != odRsp);
322 DEBUGXQ(
"Calling Finished(" <<cancel <<
')');
323 if (respWait) WakeUp();
324 if (finWait) finWait->Post();
338 Log.
Emsg(epname,
tident,
"Invalid req/rsp state; giving up on object!");
355 if (eiP != fileP->errInfo())
delete eiP;
360 if (myState == odRsp)
361 {
DEBUGXQ(
"resp sent; no additional data remains");
368 DEBUGXQ(
"wtrsp sent; resp " <<(haveResp ?
"here" :
"pend"));
375 if (!haveResp) respWait =
true;
383 int XrdSsiFileReq::Emsg(
const char *pfx,
395 if (ecode < 0) ecode = -ecode;
407 if (cbInfo) cbInfo->setErrInfo(ecode, buffer);
413 int XrdSsiFileReq::Emsg(
const char *pfx,
428 if (eNum <= 0) eNum = EFAULT;
433 snprintf(buffer,
sizeof(buffer),
"Unable to %s %s; %s", op, sessN,
eMsg);
441 if (cbInfo) cbInfo->setErrInfo(eNum, buffer);
453 bool cancel = (myState != odRsp);
458 if (alrtSent || alrtPend)
460 if (aP) aP->
next = alrtPend;
464 mHelper.
Lock(frqMutex);
472 {
case isNew:
DEBUGXQ(
"Aborting request processing");
483 case isBegun: urState = isDone;
494 case isBound: urState = isDone;
495 if (strBuff) {strBuff->Recycle(); strBuff = 0;}
496 DEBUGXQ(
"Calling Finished(" <<cancel <<
')');
497 if (respWait) WakeUp();
509 case isDone: sessN =
"bad";
518 Log.
Emsg(epname,
tident,
"Invalid req/rsp state; giving up on object!");
537 if (oucBuff)
return oucBuff->Data();
545 void XrdSsiFileReq::Init(
const char *cID)
547 tident = (cID ? strdup(cID) : strdup(
"???"));
585 EPNAME(
"ProcessResponse");
589 DEBUGXQ(
"Response presented wtr=" <<respWait);
593 if (urState != isBegun && urState != isBound)
return false;
601 DEBUGXQ(
"Resp data sz="<<Resp.blen);
606 DEBUGXQ(
"Resp err rc="<<Resp.eNum<<
" msg="<<Resp.eMsg);
611 DEBUGXQ(
"Resp file fd="<<Resp.fdnum<<
" sz="<<Resp.fsize);
631 if (respWait) WakeUp();
653 static const char *epname =
"read";
659 if (myState != doRsp)
661 return (myState == odRsp ? 0 : Emsg(epname, ENOMSG,
"read"));
668 if (respLen <= 0) {done =
true; myState = odRsp;
return 0;}
670 {memcpy(buff, Resp->buff+respOff, respLen);
671 blen = respLen; myState = odRsp; done =
true;
673 memcpy(buff, Resp->buff+respOff, blen);
674 respLen -= blen; respOff += blen;
679 cbInfo->setErrInfo(Resp->eNum, Resp->eMsg);
680 myState = odRsp; done =
true;
684 if (fileSz <= 0) {done =
true; myState = odRsp;
return 0;}
685 nbytes =
pread(Resp->fdnum, buff, blen, respOff);
688 if (!nbytes) {myState = odRsp;
return 0;}
690 return Emsg(epname, errno,
"read");
692 respOff += nbytes; fileSz -= nbytes;
697 readStrmA(Resp->strmP, buff, blen)
698 : readStrmP(Resp->strmP, buff, blen));
699 done = strmEOF && strBuff == 0;
709 return Emsg(epname, EFAULT,
"read");
719 static const char *epname =
"readStrmA";
728 {memcpy(buff, strBuff->data+respOff, blen);
729 respLen -= blen; respOff += blen;
732 memcpy(buff, strBuff->data+respOff, respLen);
734 strBuff->Recycle(); strBuff = 0;
735 blen -= respLen; buff += respLen;
738 if (!strmEOF && blen)
739 {respLen = blen; respOff = 0;
740 strBuff = strmP->
GetBuff(eObj, respLen, strmEOF);
746 if (strmEOF) {myState = odRsp;
return xlen;}
747 else if (!blen)
return xlen;
751 myState = erRsp; strmEOF =
true;
752 return Emsg(epname, eObj,
"read stream");
762 static const char *epname =
"readStrmP";
769 while(!strmEOF && (dlen = strmP->
SetBuff(eObj, buff, blen, strmEOF)) > 0)
771 if (dlen == blen)
return xlen;
772 if (dlen > blen) {eObj.
Set(0,EOVERFLOW);
break;}
773 buff += dlen; blen -= dlen;
778 if (strmEOF || !dlen) {myState = odRsp; strmEOF =
true;
return xlen;}
782 myState = erRsp; strmEOF =
true;
783 return Emsg(epname, eObj,
"read stream");
790 void XrdSsiFileReq::Recycle()
796 if (oucBuff) {oucBuff->Recycle(); oucBuff = 0;}
805 if (freeCnt >= freeMax) {aqMutex.UnLock();
delete this;}
830 if (oucBuff) {oucBuff->Recycle(); oucBuff = 0;}
841 static const char *epname =
"send";
849 if (myState != doRsp)
return 1;
856 {sfVec[1].buffer = (
char *)Resp->buff+respOff;
859 {blen = respLen; myState = odRsp;
861 respLen -= blen; respOff += blen;
870 {sfVec[1].offset = respOff; sfVec[1].
fdnum = Resp->fdnum;
872 {blen = fileSz; myState = odRsp;}
873 respOff += blen; fileSz -= blen;
878 return sendStrmA(Resp->strmP, sfDio, blen);
880 default: myState = erRsp;
881 return Emsg(epname, EFAULT,
"send");
887 if (!blen) {sfVec[1].buffer = rID; myState = odRsp;}
893 if (!rc)
return myState != odRsp;
897 rc = (rc < 0 ? EIO : EFAULT);
899 return Emsg(epname, rc,
"send");
909 static const char *epname =
"sendStrmA";
918 if (strmEOF || !(strBuff = strmP->
GetBuff(eObj, respLen, strmEOF)))
919 {myState = odRsp; strmEOF =
true;
920 if (!strmEOF) Emsg(epname, eObj,
"read stream");
928 sfVec[1].buffer = strBuff->data+respOff;
932 respLen -= blen; respOff += blen;
934 sfVec[1].
sendsz = respLen;
944 if (strBuff && !respLen) {strBuff->Recycle(); strBuff = 0;}
948 if (!rc)
return myState != odRsp;
952 rc = (rc < 0 ? EIO : EFAULT);
953 myState = erRsp; strmEOF =
true;
954 return Emsg(epname, rc,
"send");
970 if (alrtSent) {alrtSent->Recycle(); alrtSent = 0;}
974 frqMon.
Lock(frqMutex);
981 {
char hexBuff[16], binBuff[8], dotBuff[4];
983 if (!(alrtPend = alrtPend->next)) alrtLast = 0;
984 int n = alrtSent->SetInfo(eInfo, binBuff,
sizeof(binBuff));
986 DEBUGXQ(n <<
" byte alert (0x" <<
DUMPIT(binBuff, n) <<
") sent; "
987 <<(alrtPend ?
"" :
"no ") <<
"more pending");
996 if (fileP->AttnInfo(eInfo, rspP, reqID))
1009 respCB = eInfo.
getErrCB(respCBarg);
1036 {
char hexBuff[16], binBuff[8], dotBuff[4];
1037 int n = aP->
SetInfo(*wuInfo, binBuff,
sizeof(binBuff));
1039 DEBUGXQ(n <<
" byte alert (0x" <<
DUMPIT(binBuff, n) <<
") sent; "
1040 <<(alrtPend ?
"" :
"no ") <<
"more pending");
1042 if (fileP->AttnInfo(*wuInfo, rspP, reqID))
1049 respCB->Done(respCode, wuInfo, sessN);
ssize_t pread(int fildes, void *buf, size_t nbyte, off_t offset)
class XrdBuffer * XrdSfsXioHandle
static int Format(char *buff, int blen, int ecode, const char *etxt1, const char *etxt2=0)
void setErrCB(XrdOucEICB *cb, unsigned long long cbarg=0)
void Schedule(XrdJob *jp)
virtual int SendFile(int fildes)=0
static void Reclaim(XrdSfsXioHandle theHand)
static char * Buffer(XrdSfsXioHandle theHand, int *buffsz=0)
static XrdSsiAlert * Alloc(XrdSsiRespInfoMsg &aMsg)
int SetInfo(XrdOucErrInfo &eInfo, char *aMsg, int aLen)
void Set(const char *eMsg=0, int eNum=0, int eArg=0)
const std::string & Get(int &eNum) const
void Alert(XrdSsiRespInfoMsg &aMsg)
Send or receive a server generated alert.
bool WantResponse(XrdOucErrInfo &eInfo)
XrdSfsXferSize Read(bool &done, char *buffer, XrdSfsXferSize blen)
char * GetRequest(int &rLen)
bool ProcessResponse(const XrdSsiErrInfo &eInfo, const XrdSsiRespInfo &resp)
int Send(XrdSfsDio *sfDio, XrdSfsXferSize size)
static XrdSsiFileReq * Alloc(XrdOucErrInfo *eP, XrdSsiFileResource *rP, XrdSsiFileSess *fP, const char *sn, const char *id, unsigned int rnum)
void Activate(XrdOucBuffer *oP, XrdSfsXioHandle bR, int rSz)
void Done(int &Result, XrdOucErrInfo *cbInfo, const char *path=0)
void Lock(XrdSsiMutex *mutex)
static void SetMutex(XrdSsiRequest *rP, XrdSsiMutex *mP)
static XrdSsiRespInfo * RespP(XrdSsiRequest *rP)
static void onServer(XrdSsiRequest *rP)
static void CleanUp(XrdSsiRequest &reqR)
virtual void RecycleMsg(bool sent=true)=0
virtual void ProcessRequest(XrdSsiRequest &reqRef, XrdSsiResource &resRef)=0
Process a request; client-side or server-side.
virtual bool SetBuff(XrdSsiErrInfo &eRef, char *buff, int blen)
virtual Buffer * GetBuff(XrdSsiErrInfo &eRef, int &dlen, bool &last)
int Emsg(const char *esfx, int ecode, const char *text1, const char *text2=0)
int fdnum
File descriptor for data.
int sendsz
Length of data at offset.