XrdCl::XCpCtx Class Reference

#include <XrdClXCpCtx.hh>

Collaboration diagram for XrdCl::XCpCtx:

Public Member Functions
	XCpCtx (const std::vector< std::string > &urls, uint64_t blockSize, uint8_t parallelSrc, uint64_t chunkSize, uint64_t parallelChunks, int64_t fileSize)
bool	AllDone ()
void	Delete ()
std::pair< uint64_t, uint64_t >	GetBlock ()
XRootDStatus	GetChunk (XrdCl::PageInfo &ci)
bool	GetNextUrl (std::string &url)
int64_t	GetSize ()
XRootDStatus	Initialize ()
void	NotifyIdleSrc ()
void	NotifyInitExpectant ()
void	PutChunk (PageInfo *chunk)
void	RemoveSrc (XCpSrc *src)
XCpCtx *	Self ()
void	SetFileSize (int64_t size)
XCpSrc *	WeakestLink (XCpSrc *exclude)

Detailed Description

Definition at line 40 of file XrdClXCpCtx.hh.

Constructor & Destructor Documentation

XCpCtx()

XrdCl::XCpCtx::XCpCtx	(	const std::vector< std::string > &	urls,
		uint64_t	blockSize,
		uint8_t	parallelSrc,
		uint64_t	chunkSize,
		uint64_t	parallelChunks,
		int64_t	fileSize
	)

Constructor

Parameters

urls	: list of replica urls
blockSize	: the default block size
parallelSrc	: maximum number of parallel sources
chunkSize	: the default chunk size
parallelChunks	: the default number of parallel chunks per source
fileSize	: the file size if specified in the metalink file (-1 indicates that the file size is not known and a stat should be done)

Definition at line 36 of file XrdClXCpCtx.cc.

                                                                                                                                                         :
      pUrls( std::deque<std::string>( urls.begin(), urls.end() ) ), pBlockSize( blockSize ),
      pParallelSrc( parallelSrc ), pChunkSize( chunkSize ), pParallelChunks( parallelChunks ),
      pOffset( 0 ), pFileSize( -1 ), pFileSizeCV( 0 ), pDataReceived( 0 ), pDone( false ),
      pDoneCV( 0 ), pRefCount( 1 )
{
  SetFileSize( fileSize );
}

References SetFileSize().

Here is the call graph for this function:

Member Function Documentation

AllDone()

bool XrdCl::XCpCtx::AllDone

(

)

Returns true if all chunks have been transferred, otherwise blocks until NotifyIdleSrc is called, or a 1 minute timeout occurs.

Returns

: true is all chunks have been transferred, false otherwise.

Definition at line 177 of file XrdClXCpCtx.cc.

{
  XrdSysCondVarHelper lck( pDoneCV );
 
  if( !pDone )
    pDoneCV.Wait( 60 );
 
  return pDone;
}

References XrdSysCondVar::Wait().

Here is the call graph for this function:

Delete()

void XrdCl::XCpCtx::Delete ( )

inline

Deletes the instance if the reference counter reached 0.

Definition at line 61 of file XrdClXCpCtx.hh.

    {
      XrdSysMutexHelper lck( pMtx );
      --pRefCount;
      if( !pRefCount )
      {
        lck.UnLock();
        delete this;
      }
    }

References XrdSysMutexHelper::UnLock().

Referenced by XrdCl::XCpSrc::Start().

Here is the call graph for this function:

Here is the caller graph for this function:

GetBlock()

std::pair< uint64_t, uint64_t > XrdCl::XCpCtx::GetBlock

(

)

Get next block that has to be transferred

Returns

: pair of offset and block size

Definition at line 92 of file XrdClXCpCtx.cc.

{
  XrdSysMutexHelper lck( pMtx );
 
  uint64_t blkSize = pBlockSize, offset = pOffset;
  if( pOffset + blkSize > uint64_t( pFileSize ) )
    blkSize = pFileSize - pOffset;
  pOffset += blkSize;
 
  return std::make_pair( offset, blkSize );
}

GetChunk()

XRootDStatus XrdCl::XCpCtx::GetChunk

(

XrdCl::PageInfo &

ci

)

Gets the next chunk from the sink, if the sink is empty blocks.

Parameters

ci	: the chunk retrieved from sink (output parameter)

Returns

: stError if we failed to transfer the file, stOK otherwise, with one of the following codes:

• suDone : the whole file has been transferred, we are done
• suContinue : a chunk has been written into ci, continue calling GetChunk in order to retrieve remaining chunks
• suRetry : a chunk has not been written into ci, try again.

Definition at line 140 of file XrdClXCpCtx.cc.

{
  // if we received all the data we are done here
  if( pDataReceived == uint64_t( pFileSize ) )
  {
    XrdSysCondVarHelper lck( pDoneCV );
    pDone = true;
    pDoneCV.Broadcast();
    return XRootDStatus( stOK, suDone );
  }
 
  // if we don't have active sources it means we failed
  if( GetRunning() == 0 )
  {
    XrdSysCondVarHelper lck( pDoneCV );
    pDone = true;
    pDoneCV.Broadcast();
    return XRootDStatus( stError, errNoMoreReplicas );
  }
 
  PageInfo *chunk = pSink.Get();
  if( chunk )
  {
    pDataReceived += chunk->GetLength();
    ci = std::move( *chunk );
    delete chunk;
    return XRootDStatus( stOK, suContinue );
  }
 
  return XRootDStatus( stOK, suRetry );
}

References XrdSysCondVar::Broadcast(), XrdCl::errNoMoreReplicas, XrdCl::PageInfo::GetLength(), XrdCl::stError, XrdCl::stOK, XrdCl::suContinue, XrdCl::suDone, and XrdCl::suRetry.

Here is the call graph for this function:

GetNextUrl()

bool XrdCl::XCpCtx::GetNextUrl

(

std::string &

url

)

Gets the next URL from the list of file replicas

Parameters

url	: the output parameter

Returns

: true if a url has been written to the url parameter, false otherwise

Definition at line 57 of file XrdClXCpCtx.cc.

{
  XrdSysMutexHelper lck( pMtx );
  if( pUrls.empty() ) return false;
  url = pUrls.front();
  pUrls.pop();
  return true;
}

GetSize()

int64_t XrdCl::XCpCtx::GetSize ( )

inline

Get file size. The call blocks until the file size is being set using SetFileSize.

Definition at line 129 of file XrdClXCpCtx.hh.

    {
      XrdSysCondVarHelper lck( pFileSizeCV );
      while( pFileSize < 0 && GetRunning() > 0 ) pFileSizeCV.Wait();
      return pFileSize;
    }

References XrdSysCondVar::Wait().

Here is the call graph for this function:

Initialize()

XRootDStatus XrdCl::XCpCtx::Initialize

(

)

Starts one thread per source, each thread tries to open a file, stat the file if necessary, and then starts reading the file, all chunks read go to the sink.

Returns

Error if we were not able to create any threads

Definition at line 121 of file XrdClXCpCtx.cc.

{
  for( uint8_t i = 0; i < pParallelSrc; ++i )
  {
    XCpSrc *src = new XCpSrc( pChunkSize, pParallelChunks, pFileSize, this );
    pSources.push_back( src );
    src->Start();
  }
 
  if( pSources.empty() )
  {
    Log *log = DefaultEnv::GetLog();
    log->Error( UtilityMsg, "Failed to initialize (failed to create new threads)" );
    return XRootDStatus( stError, errInternal, EAGAIN, "XCpCtx: failed to create new threads." );
  }
 
  return XRootDStatus();
}

References XrdCl::errInternal, XrdCl::Log::Error(), XrdCl::DefaultEnv::GetLog(), XrdCl::XCpSrc::Start(), XrdCl::stError, and XrdCl::UtilityMsg.

Here is the call graph for this function:

NotifyIdleSrc()

void XrdCl::XCpCtx::NotifyIdleSrc

(

)

Notify idle sources, used in two case:

• if one of the sources failed and an idle source needs to take over
• or if we are done and all idle source should be stopped

Definition at line 172 of file XrdClXCpCtx.cc.

{
  pDoneCV.Broadcast();
}

References XrdSysCondVar::Broadcast().

Here is the call graph for this function:

NotifyInitExpectant()

void XrdCl::XCpCtx::NotifyInitExpectant ( )

inline

Notify those who are waiting for initialization. In particular the GetSize() caller will be waiting on the result of initialization.

Definition at line 197 of file XrdClXCpCtx.hh.

    {
      pFileSizeCV.Broadcast();
    }

References XrdSysCondVar::Broadcast().

Here is the call graph for this function:

PutChunk()

void XrdCl::XCpCtx::PutChunk

(

PageInfo *

chunk

)

Put a chunk into the sink

Parameters

chunk

: the chunk

Definition at line 87 of file XrdClXCpCtx.cc.

{
  pSink.Put( chunk );
}

RemoveSrc()

void XrdCl::XCpCtx::RemoveSrc ( XCpSrc *  src )

inline

Remove given source

Parameters

src	: the source to be removed

Definition at line 167 of file XrdClXCpCtx.hh.

    {
      XrdSysMutexHelper lck( pMtx );
      pSources.remove( src );
    }

Referenced by XrdCl::XCpSrc::Start().

Here is the caller graph for this function:

Self()

XCpCtx * XrdCl::XCpCtx::Self ( )

inline

Increments the reference counter.

Returns

: myself.

Definition at line 77 of file XrdClXCpCtx.hh.

    {
      XrdSysMutexHelper lck( pMtx );
      ++pRefCount;
      return this;
    }

SetFileSize()

void XrdCl::XCpCtx::SetFileSize

(

int64_t

size

)

Set the file size (GetSize will block until SetFileSize will be called). Also calculates the block size.

Parameters

size	: file size

Definition at line 104 of file XrdClXCpCtx.cc.

{
  XrdSysMutexHelper lck( pMtx );
  if( pFileSize < 0 && size >= 0 )
  {
    XrdSysCondVarHelper lck( pFileSizeCV );
    pFileSize = size;
    pFileSizeCV.Broadcast();
 
    if( pBlockSize > uint64_t( pFileSize ) / pParallelSrc )
      pBlockSize = pFileSize / pParallelSrc;
 
    if( pBlockSize < pChunkSize )
      pBlockSize = pChunkSize;
  }
}

References XrdSysCondVar::Broadcast().

Referenced by XCpCtx().

Here is the call graph for this function:

Here is the caller graph for this function:

WeakestLink()

XCpSrc * XrdCl::XCpCtx::WeakestLink

(

XCpSrc *

exclude

)

Get the 'weakest' sources

Parameters

exclude

: the source that is excluded from the search

Returns

: the weakest source

Definition at line 66 of file XrdClXCpCtx.cc.

{
  uint64_t transferRate = -1; // set transferRate to max uint64 value
  XCpSrc *ret = 0;
 
  std::list<XCpSrc*>::iterator itr;
  for( itr = pSources.begin() ; itr != pSources.end() ; ++itr )
  {
    XCpSrc *src = *itr;
    if( src == exclude ) continue;
    uint64_t tmp = src->TransferRate();
    if( src->HasData() && tmp < transferRate )
    {
      ret = src;
      transferRate = tmp;
    }
  }
 
  return ret;
}

References XrdCl::XCpSrc::HasData(), and XrdCl::XCpSrc::TransferRate().

Here is the call graph for this function:

---

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

• XrdClXCpCtx.hh
• XrdClXCpCtx.cc