#include <XrdXrootdAioTask.hh>

Inheritance diagram for XrdXrootdAioTask:

Collaboration diagram for XrdXrootdAioTask:

Public Member Functions
void	Completed (XrdXrootdAioBuff *aioP)

const char *	ID ()

void	Init (XrdXrootdProtocol protP, XrdXrootdResponse &resp, XrdXrootdFile fP)

virtual void	Read (long long offs, int dlen)=0

virtual void	Recycle (bool release)=0

XrdXrootdProtocol *	urProtocol ()

virtual int	Write (long long offs, int dlen)=0

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

virtual	~XrdJob ()

virtual void	DoIt ()=0

Protected Member Functions
	XrdXrootdAioTask (const char *what="aio request")

virtual	~XrdXrootdAioTask ()

virtual void	CopyF2L ()=0

virtual int	CopyL2F ()=0

virtual bool	CopyL2F (XrdXrootdAioBuff *aioP)=0

bool	Drain ()

int	gdDone () override

void	gdFail () override

XrdXrootdAioBuff *	getBuff (bool wait)

void	SendError (int rc, const char *eText)

void	SendFSError (int rc)

bool	Validate (XrdXrootdAioBuff *aioP)

Protected Attributes
union {
XrdXrootdNormAio *	nextNorm

XrdXrootdPgrwAio *	nextPgrw

XrdXrootdAioTask *	nextTask

};

union {
XrdXrootdAioBuff *	finalRead

XrdXrootdAioBuff *	pendWrite

};

XrdSysMutex	aioMutex

XrdSysCondVar2	aioReady

char	aioState

XrdXrootdFile *	dataFile

int	dataLen

XrdLink *	dataLink

off_t	dataOffset

off_t	highOffset

RAtomic_uchar	inFlight

RAtomic_bool	isDone

XrdXrootdAioBuff *	pendQ

XrdXrootdAioBuff *	pendQEnd

XrdXrootdProtocol *	Protocol

XrdXrootdResponse	Response

char	Status

Static Protected Attributes
static const int	aioDead = 0x01

static const int	aioHeld = 0x02

static const int	aioPage = 0x04

static const int	aioRead = 0x08

static const int	aioSchd = 0x10

static const int	Offline = 0

static const int	Running = 1

static const char *	TraceID = "AioTask"

static const int	Waiting = 2

Friends
class	XrdXrootdAioFob

Additional Inherited Members
Public Attributes inherited from XrdJob
const char *	Comment

XrdJob *	NextJob

Detailed Description

Definition at line 46 of file XrdXrootdAioTask.hh.

Constructor & Destructor Documentation

◆ XrdXrootdAioTask()

XrdXrootdAioTask::XrdXrootdAioTask ( const char * what = "aio request" )

inlineprotected

Definition at line 69 of file XrdXrootdAioTask.hh.

70 : XrdJob(what), aioReady(aioMutex) {}

XrdJob::XrdJob

XrdJob(const char *desc="")

Definition: XrdJob.hh:51

XrdXrootdAioTask::aioReady

XrdSysCondVar2 aioReady

Definition: XrdXrootdAioTask.hh:87

XrdXrootdAioTask::aioMutex

XrdSysMutex aioMutex

Definition: XrdXrootdAioTask.hh:86

◆ ~XrdXrootdAioTask()

virtual XrdXrootdAioTask::~XrdXrootdAioTask ( )

inlineprotectedvirtual

Definition at line 71 of file XrdXrootdAioTask.hh.

71 {}

Member Function Documentation

◆ Completed()

void XrdXrootdAioTask::Completed ( XrdXrootdAioBuff * aioP )

Definition at line 73 of file XrdXrootdAioTask.cc.

 {
 // Lock this code path
 //
    aioMutex.Lock();
  
 // If this request is not running and completed then take a shortcut.
 //
    if (Status == Offline && isDone)
       {aioP->Recycle();
        inFlight--;
        aioMutex.UnLock();
        if (inFlight <= 0) Recycle(true);
        return;
       }
  
 // Add this element to the end of the queue
 //
    aioP->next = 0;
    if (!pendQ) pendQEnd = pendQ = aioP;
       else {pendQEnd->next = aioP;
             pendQEnd = aioP;
            }
  
 // Check if the request is waiting for our buffer tell it now has one. Otherwise,
 // if the task is offline then it cannot be done (see above); so schedule it.
 //
    if (Status != Running)
       {if (Status == Waiting) aioReady.Signal();
           else Sched->Schedule(this);
        Status = Running;
       }
  
    aioMutex.UnLock();
 }

References XrdXrootdAioBuff::next, XrdXrootdAioBuff::Recycle(), XrdXrootd::Sched, and XrdScheduler::Schedule().

Here is the call graph for this function:

◆ CopyF2L()

virtual void XrdXrootdAioTask::CopyF2L ( )

protectedpure virtual

◆ CopyL2F() [1/2]

virtual int XrdXrootdAioTask::CopyL2F ( )

protectedpure virtual

◆ CopyL2F() [2/2]

virtual bool XrdXrootdAioTask::CopyL2F ( XrdXrootdAioBuff * aioP )

protectedpure virtual

◆ Drain()

bool XrdXrootdAioTask::Drain ( )

protected

Definition at line 113 of file XrdXrootdAioTask.cc.

 {
    XrdXrootdAioBuff *aioP;
    int maxWait = 6;          // Max seconds to wait for outstanding requests
  
 // Reap as many aio object as you can
 //
    aioMutex.Lock();
    while(inFlight > 0)
         {while((aioP = pendQ))
             {if (!(pendQ = aioP->next)) pendQEnd = 0;
              aioMutex.UnLock(); // Open a window of opportunity
              inFlight--;
              aioP->Recycle();
              aioMutex.Lock();
             }
          if (inFlight <= 0 || !Wait4Buff(maxWait)) break;
         }
  
 // If there are still in flight requets, issue message and we will run the
 // drain in the background.
 //
    if (inFlight > 0)
       {char buff[128];
        snprintf(buff, sizeof(buff),
                       "aio%c overdue %d inflight request%s for",
                       (aioState & aioRead ? 'R' : 'W'), int(inFlight),
                       (inFlight > 1 ? "s" : ""));
        eLog.Emsg("AioTask", buff, dataLink->ID, dataFile->FileKey);
       }
  
 // Indicate we are going offline and tell the caller if we need to stay
 // alive to drain the tardy requests in the background.
 //
    Status = Offline;
    isDone = true;
    aioMutex.UnLock();
    return inFlight <= 0;
 }

References XrdXrootd::eLog, XrdSysError::Emsg(), XrdXrootdAioBuff::next, and XrdXrootdAioBuff::Recycle().

Here is the call graph for this function:

◆ gdDone()

int XrdXrootdAioTask::gdDone ( )

overrideprotected

Definition at line 157 of file XrdXrootdAioTask.cc.

 {
    XrdXrootdAioBuff *bP = pendWrite;
    int rc;
  
 // Do some debugging
 //
    TRACEP(DEBUG,"gdDone: "<<(void *)this<<" pendWrite "
                 <<(pendWrite != 0 ? "set":"not set"));
  
 // This is a callback indicating the pending aio object has all of the data.
 // Resume sending data to the destination.
 //
    pendWrite = 0;
    if (!bP) rc = CopyL2F();
       else {if (CopyL2F(bP) && (inFlight || !isDone)) rc = CopyL2F();
                else rc = 0;
            }
  
 // Do some debugging
 //
    TRACEP(DEBUG,"gdDone: "<<(void *)this<<" ending rc="<<rc);
  
 // If we are not pausing for data to be delivered. Drain any oustanding aio
 // requests and discard left over bytes, if any. Note we must copy the left
 // over length as we may recycle before discarding as discard must be last.
 //
    if (rc <= 0)
       {XrdXrootdProtocol* prot = Protocol;
        int dlen = dataLen;
        if (!inFlight) Recycle(true);
           else Recycle(Drain());
        if (!rc && dlen) return prot->getDump(Comment, dlen);
       }
    return rc;
 }

References DEBUG, XrdXrootdProtocol::getDump(), and TRACEP.

Here is the call graph for this function:

◆ gdFail()

void XrdXrootdAioTask::gdFail ( )

overrideprotected

Definition at line 198 of file XrdXrootdAioTask.cc.

 {
    char eBuff[512];
  
 // Do some tracing
 //
    TRACEP(DEBUG,"gdFail: "<<(void *)this);
  
 // Format message for display
 //
    snprintf(eBuff, sizeof(eBuff), "link error aborted %s for", Comment);
    eLog.Emsg("AioTask", eBuff, dataLink->ID, dataFile->FileKey);
  
 // This is a callback indicating the link is dead. Terminate this operation.
 //
    isDone = true;
    aioState |= aioDead;
    dataLen = 0;
    if (pendWrite) {pendWrite->Recycle(); pendWrite = 0;}
  
 // If this is a read, cancel all queued read requests
 //
    if (aioState & aioRead) dataFile->aioFob->Reset(Protocol);
  
 // If we still have any requests in flight drain them.
 //
    if (!inFlight) Recycle(true);
       else Recycle(Drain());
 }

References DEBUG, XrdXrootd::eLog, XrdSysError::Emsg(), and TRACEP.

Here is the call graph for this function:

◆ getBuff()

XrdXrootdAioBuff * XrdXrootdAioTask::getBuff ( bool wait )

protected

Definition at line 232 of file XrdXrootdAioTask.cc.

 {
    XrdXrootdAioBuff* aioP;
  
 // Try to get the next buffer
 //
    aioMutex.Lock();
 do{if ((aioP = pendQ))
       {if (!(pendQ = aioP->next)) pendQEnd = 0;
        aioMutex.UnLock();
        inFlight--;
        return aioP;
       }
  
 // If the caller does not want to wait or if there is nothing in flight, return
 //
    if (!wait || !inFlight)
       {aioMutex.UnLock();
        return 0;
       }
  
 // So, wait for a buffer to arrive
 //
   } while(Wait4Buff());
  
 // We timed out and this is considered an error
 //
    aioMutex.UnLock();
    SendError(ETIMEDOUT, (aioState & aioRead ? "aio file read timed out"
                                             : "aio file write timed out"));
    return 0;
 }

References XrdXrootdAioBuff::next.

◆ ID()

const char * XrdXrootdAioTask::ID ( void )

Definition at line 269 of file XrdXrootdAioTask.cc.

269 {return dataLink->ID;}

◆ Init()

void XrdXrootdAioTask::Init	(	XrdXrootdProtocol *	protP,
		XrdXrootdResponse &	resp,
		XrdXrootdFile *	fP
	)

Definition at line 275 of file XrdXrootdAioTask.cc.

 {
  
 // Reset the object
 //
    pendQEnd   = pendQ = 0;
    finalRead  = 0; // Also sets pendWrite
    Protocol   = protP;
    dataLink   = resp.theLink();
    Response   = resp;
    dataFile   = fP;
    aioState   = 0;
    inFlight   = 0;
    isDone     = false;
    Status     = Running;
 }

References XrdXrootdResponse::theLink().

Referenced by XrdXrootdNormAio::Alloc(), and XrdXrootdPgrwAio::Alloc().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ Read()

virtual void XrdXrootdAioTask::Read	(	long long	offs,
		int	dlen
	)

pure virtual

Implemented in XrdXrootdPgrwAio, and XrdXrootdNormAio.

◆ Recycle()

virtual void XrdXrootdAioTask::Recycle ( bool release )

pure virtual

Implemented in XrdXrootdPgrwAio, and XrdXrootdNormAio.

Referenced by XrdXrootdAioFob::Reset().

Here is the caller graph for this function:

◆ SendError()

void XrdXrootdAioTask::SendError	(	int	rc,
		const char *	eText
	)

protected

Definition at line 298 of file XrdXrootdAioTask.cc.

 {
    char eBuff[1024];
  
 // If there is no error text, use the rc
 //
    if (!eText) eText = (rc ? XrdSysE2T(rc) : "invalid error code");
  
 // For message for display
 //
    snprintf(eBuff, sizeof(eBuff), "async %s failed for %s;",
             (aioState & aioRead ? "read" : "write"), dataLink->ID);
    eLog.Emsg("AioTask", eBuff, eText, dataFile->FileKey);
  
 // If this request is still active, send the error to the client
 //
    if (!isDone)
       {XErrorCode eCode = (XErrorCode)XProtocol::mapError(rc);
        if (Response.Send(eCode, eText))
           {aioState |= aioDead;
            dataLen = 0;
           } else if (aioState & aioRead) dataLen = 0;
        isDone = true;
       }
 }

References XrdXrootd::eLog, XrdSysError::Emsg(), XProtocol::mapError(), and XrdSysE2T().

Here is the call graph for this function:

◆ SendFSError()

void XrdXrootdAioTask::SendFSError ( int rc )

protected

Definition at line 328 of file XrdXrootdAioTask.cc.

 {
    XrdOucErrInfo &myError = dataFile->XrdSfsp->error;
    int eCode;
  
 // We can only handle actual errors. Under some conditions a redirect (e.g.
 // Xcache) can return other error codes. We treat these as server errors.
 //
    if (rc != SFS_ERROR)
       {char eBuff[256];
        snprintf(eBuff, sizeof(eBuff), "fs returned unexpected rc %d", rc);
        SendError(EFAULT, eBuff);
        if (myError.extData()) myError.Reset();
        return;
       }
  
 // Handle file system error but only if we are still alive
 //
    if (!isDone)
       {const char *eMsg = myError.getErrText(eCode);
        eLog.Emsg("AioTask", dataLink->ID, eMsg, dataFile->FileKey);
        int rc = XProtocol::mapError(eCode);
        if (Response.Send((XErrorCode)rc, eMsg))
           {aioState |= aioDead;
            dataLen = 0;
           } else if (aioState & aioRead) dataLen = 0;
        isDone = true;
       }
  
 // Clear error message and recycle aio object if need be
 //
    if (myError.extData()) myError.Reset();
 }

References XrdXrootd::eLog, XrdSysError::Emsg(), eMsg, XrdOucErrInfo::extData(), XrdOucErrInfo::getErrText(), XProtocol::mapError(), XrdOucErrInfo::Reset(), and SFS_ERROR.

Here is the call graph for this function:

◆ urProtocol()

XrdXrootdProtocol* XrdXrootdAioTask::urProtocol ( )

inline

Definition at line 63 of file XrdXrootdAioTask.hh.

63 {return Protocol;}

Referenced by XrdXrootdAioBuff::Alloc(), XrdXrootdAioPgrw::Alloc(), and XrdXrootdAioPgrw::Recycle().

Here is the caller graph for this function:

◆ Validate()

bool XrdXrootdAioTask::Validate ( XrdXrootdAioBuff * aioP )

protected

Definition at line 366 of file XrdXrootdAioTask.cc.

 {
    ssize_t aioResult = aioP->Result;
    off_t   aioOffset = aioP->sfsAio.aio_offset;
    int     aioLength = aioP->sfsAio.aio_nbytes;
  
 // Step 1: Check if this request is already completed. This may be the case
 //         if we had a previous error.
 //
    if (isDone) return false;
  
 // Step 2: Check if an error occurred as this will terminate the request even
 //         if we have not sent all of the data.
 //
    if (aioP->Result < 0)
       {SendError(-aioP->Result, 0);
        return false;
       }
  
 // Step 3: Check for a short read which signals that no more data past this
 //         offset is forthcomming. Save it as we will send a final response
 //         using this element. We discard zero length reads. It's an error if we
 //         get more than one short read with data or if its offset is less than
 //         the highest full read element.
 //
    if (aioResult < aioLength)
       {dataLen = 0;
        if (!aioResult)
           {if ((finalRead && aioOffset < finalRead->sfsAio.aio_offset)
            ||  aioOffset < highOffset) SendError(EFAULT, "embedded null block");
            return false;
           } else {
            if (aioOffset < highOffset)
               {SendError(ENODEV, "embedded short block");
                return false;
               } else {
                if (finalRead) SendError(ENODEV, "multiple short blocks");
                   else {finalRead = aioP;
                         highOffset = aioOffset;
                        }
               }
           }
        return false;
       }
  
 // Step 4: This is a full read and its offset must be lower than the offset of
 //         any short read we have encountered.
 //
    if (finalRead && aioOffset >= finalRead->sfsAio.aio_offset)
       {SendError(ENODEV, "read offset past EOD");
        return false;
       }
    if (aioOffset > highOffset) highOffset = aioOffset;
    return true;
 }

References aiocb::aio_nbytes, aiocb::aio_offset, XrdSfsAio::Result, and XrdSfsAio::sfsAio.

◆ Write()

virtual int XrdXrootdAioTask::Write	(	long long	offs,
		int	dlen
	)

pure virtual

Implemented in XrdXrootdPgrwAio, and XrdXrootdNormAio.

Friends And Related Function Documentation

◆ XrdXrootdAioFob

friend class XrdXrootdAioFob

friend

Definition at line 49 of file XrdXrootdAioTask.hh.

Member Data Documentation

◆

union { ... }

◆

union { ... }

◆ aioDead

const int XrdXrootdAioTask::aioDead = 0x01

staticprotected

Definition at line 115 of file XrdXrootdAioTask.hh.

◆ aioHeld

const int XrdXrootdAioTask::aioHeld = 0x02

staticprotected

Definition at line 116 of file XrdXrootdAioTask.hh.

◆ aioMutex

XrdSysMutex XrdXrootdAioTask::aioMutex

protected

Definition at line 86 of file XrdXrootdAioTask.hh.

◆ aioPage

const int XrdXrootdAioTask::aioPage = 0x04

staticprotected

Definition at line 117 of file XrdXrootdAioTask.hh.

◆ aioRead

const int XrdXrootdAioTask::aioRead = 0x08

staticprotected

Definition at line 118 of file XrdXrootdAioTask.hh.

◆ aioReady

XrdSysCondVar2 XrdXrootdAioTask::aioReady

protected

Definition at line 87 of file XrdXrootdAioTask.hh.

◆ aioSchd

const int XrdXrootdAioTask::aioSchd = 0x10

staticprotected

Definition at line 119 of file XrdXrootdAioTask.hh.

◆ aioState

char XrdXrootdAioTask::aioState

protected

Definition at line 106 of file XrdXrootdAioTask.hh.

◆ dataFile

XrdXrootdFile* XrdXrootdAioTask::dataFile

protected

Definition at line 98 of file XrdXrootdAioTask.hh.

◆ dataLen

int XrdXrootdAioTask::dataLen

protected

Definition at line 104 of file XrdXrootdAioTask.hh.

◆ dataLink

XrdLink* XrdXrootdAioTask::dataLink

protected

Definition at line 97 of file XrdXrootdAioTask.hh.

◆ dataOffset

off_t XrdXrootdAioTask::dataOffset

protected

Definition at line 103 of file XrdXrootdAioTask.hh.

◆ highOffset

off_t XrdXrootdAioTask::highOffset

protected

Definition at line 102 of file XrdXrootdAioTask.hh.

◆ inFlight

RAtomic_uchar XrdXrootdAioTask::inFlight

protected

Definition at line 107 of file XrdXrootdAioTask.hh.

◆ isDone

RAtomic_bool XrdXrootdAioTask::isDone

protected

Definition at line 108 of file XrdXrootdAioTask.hh.

◆ Offline

const int XrdXrootdAioTask::Offline = 0

staticprotected

Definition at line 123 of file XrdXrootdAioTask.hh.

◆ pendQ

XrdXrootdAioBuff* XrdXrootdAioTask::pendQ

protected

Definition at line 88 of file XrdXrootdAioTask.hh.

◆ pendQEnd

XrdXrootdAioBuff* XrdXrootdAioTask::pendQEnd

protected

Definition at line 89 of file XrdXrootdAioTask.hh.

◆ Protocol

XrdXrootdProtocol* XrdXrootdAioTask::Protocol

protected

Definition at line 96 of file XrdXrootdAioTask.hh.

Referenced by XrdXrootdAioFob::Schedule().

◆ Response

XrdXrootdResponse XrdXrootdAioTask::Response

protected

Definition at line 111 of file XrdXrootdAioTask.hh.

◆ Running

const int XrdXrootdAioTask::Running = 1

staticprotected

Definition at line 124 of file XrdXrootdAioTask.hh.

◆ Status

char XrdXrootdAioTask::Status

protected

Definition at line 109 of file XrdXrootdAioTask.hh.

◆ TraceID

const char * XrdXrootdAioTask::TraceID = "AioTask"

staticprotected

Definition at line 84 of file XrdXrootdAioTask.hh.

◆ Waiting

const int XrdXrootdAioTask::Waiting = 2

staticprotected

Definition at line 125 of file XrdXrootdAioTask.hh.

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

Public Member Functions

Protected Member Functions

Protected Attributes

Static Protected Attributes

Friends

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ XrdXrootdAioTask()

◆ ~XrdXrootdAioTask()

Member Function Documentation

◆ Completed()

◆ CopyF2L()

◆ CopyL2F() [1/2]

◆ CopyL2F() [2/2]

◆ Drain()

◆ gdDone()

◆ gdFail()

◆ getBuff()

◆ ID()

◆ Init()

◆ Read()

◆ Recycle()

◆ SendError()

◆ SendFSError()

◆ urProtocol()

◆ Validate()

◆ Write()

Friends And Related Function Documentation

◆ XrdXrootdAioFob

Member Data Documentation

◆

◆

◆ aioDead

◆ aioHeld

◆ aioMutex

◆ aioPage

◆ aioRead

◆ aioReady

◆ aioSchd

◆ aioState

◆ dataFile

◆ dataLen

◆ dataLink

◆ dataOffset

◆ highOffset

◆ inFlight

◆ isDone

◆ Offline

◆ pendQ

◆ pendQEnd

◆ Protocol

◆ Response

◆ Running

◆ Status

◆ TraceID

◆ Waiting