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	Release ()
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 ), pDeleteCV( 0 ), pDelete( false )
{
  SetFileSize( fileSize );
}

References SetFileSize().

Referenced by Self().

Here is the call graph for this function:

Here is the caller 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 193 of file XrdClXCpCtx.cc.

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

Delete()

void XrdCl::XCpCtx::Delete ( )

inline

Decrements the reference count and then waits for it to reach zero, then deletes the instance. Should only be called once.

Definition at line 62 of file XrdClXCpCtx.hh.

    {
      XrdSysMutexHelper lckmtx( pMtx );
      --pRefCount;
      if( !pRefCount )
      {
        lckmtx.UnLock();
        delete this;
        return;
      }
      lckmtx.UnLock();
 
      XrdSysCondVarHelper lckcv( pDoneCV );
      pDone = true;
      pDoneCV.Broadcast();
      lckcv.UnLock();
 
      lckcv.Lock( &pDeleteCV );
      while( !pDelete ) pDeleteCV.Wait();
      lckcv.UnLock();
      delete this;
    }

References XrdSysCondVarHelper::Lock(), XrdSysCondVarHelper::UnLock(), and XrdSysMutexHelper::UnLock().

Here is the call 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 95 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 156 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 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 157 of file XrdClXCpCtx.hh.

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

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 124 of file XrdClXCpCtx.cc.

{
  for( uint8_t i = 0; i < pParallelSrc; ++i )
  {
    XCpSrc *src = new XCpSrc( pChunkSize, pParallelChunks, pFileSize, this );
    pSources.push_back( src );
  }
 
  auto scpy = pSources;
  bool ok   = false;
  for(auto src: scpy) {
    if( src->Start() )
    {
      // src destructor will remove src from pSources
      src->Delete();
    }
    else
    {
      ok = true;
    }
  }
 
  if( !ok )
  {
    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::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 188 of file XrdClXCpCtx.cc.

{
  pDoneCV.Broadcast();
}

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 225 of file XrdClXCpCtx.hh.

    {
      pFileSizeCV.Broadcast();
    }

PutChunk()

void XrdCl::XCpCtx::PutChunk

(

PageInfo *

chunk

)

Put a chunk into the sink

Parameters

chunk

: the chunk

Definition at line 90 of file XrdClXCpCtx.cc.

{
  pSink.Put( chunk );
}

Release()

void XrdCl::XCpCtx::Release ( )

inline

Decrements the reference count and signal when we reach 0

Definition at line 88 of file XrdClXCpCtx.hh.

    {
      XrdSysMutexHelper lck( pMtx );
      --pRefCount;
      if( !pRefCount )
      {
        XrdSysCondVarHelper lckcv( pDeleteCV );
        pDelete = true;
        pDeleteCV.Broadcast();
      }
    }

RemoveSrc()

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

inline

Remove given source

Parameters

src	: the source to be removed

Definition at line 195 of file XrdClXCpCtx.hh.

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

Self()

XCpCtx * XrdCl::XCpCtx::Self ( )

inline

Increments the reference counter.

Returns

: myself.

Definition at line 105 of file XrdClXCpCtx.hh.

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

References XCpCtx().

Here is the call graph for this function:

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 107 of file XrdClXCpCtx.cc.

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

Referenced by XCpCtx().

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;
  XrdSysMutexHelper lck( pMtx );
 
  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;
    }
  }
 
  if( !ret ) return ret;
  return ret->Self();
}

References XrdCl::XCpSrc::HasData(), XrdCl::XCpSrc::Self(), 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