36 XCpCtx::XCpCtx(
const std::vector<std::string> &urls, uint64_t blockSize, uint8_t parallelSrc, uint64_t chunkSize, uint64_t parallelChunks, int64_t fileSize ) :
37 pUrls( std::deque<std::string>( urls.begin(), urls.end() ) ), pBlockSize( blockSize ),
38 pParallelSrc( parallelSrc ), pChunkSize( chunkSize ), pParallelChunks( parallelChunks ),
39 pOffset( 0 ), pFileSize( -1 ), pFileSizeCV( 0 ), pDataReceived( 0 ), pDone( false ),
40 pDoneCV( 0 ), pRefCount( 1 )
49 while( !pSink.IsEmpty() )
60 if( pUrls.empty() )
return false;
68 uint64_t transferRate = -1;
71 std::list<XCpSrc*>::iterator itr;
72 for( itr = pSources.begin() ; itr != pSources.end() ; ++itr )
75 if( src == exclude )
continue;
77 if( src->
HasData() && tmp < transferRate )
96 uint64_t blkSize = pBlockSize, offset = pOffset;
97 if( pOffset + blkSize > uint64_t( pFileSize ) )
98 blkSize = pFileSize - pOffset;
101 return std::make_pair( offset, blkSize );
107 if( pFileSize < 0 && size >= 0 )
113 if( pBlockSize > uint64_t( pFileSize ) / pParallelSrc )
114 pBlockSize = pFileSize / pParallelSrc;
116 if( pBlockSize < pChunkSize )
117 pBlockSize = pChunkSize;
123 for( uint8_t i = 0; i < pParallelSrc; ++i )
125 XCpSrc *src =
new XCpSrc( pChunkSize, pParallelChunks, pFileSize,
this );
126 pSources.push_back( src );
130 if( pSources.empty() )
133 log->
Error(
UtilityMsg,
"Failed to initialize (failed to create new threads)" );
143 if( pDataReceived == uint64_t( pFileSize ) )
152 if( GetRunning() == 0 )
164 ci = std::move( *chunk );
187 size_t XCpCtx::GetRunning()
190 size_t nbRunning = 0;
191 std::list<XCpSrc*>::iterator itr;
192 for( itr = pSources.begin() ; itr != pSources.end() ; ++ itr)
193 if( (*itr)->IsRunning() )
static Log * GetLog()
Get default log.
void Error(uint64_t topic, const char *format,...)
Report an error.
bool GetNextUrl(std::string &url)
XCpSrc * WeakestLink(XCpSrc *exclude)
void PutChunk(PageInfo *chunk)
XCpCtx(const std::vector< std::string > &urls, uint64_t blockSize, uint8_t parallelSrc, uint64_t chunkSize, uint64_t parallelChunks, int64_t fileSize)
void SetFileSize(int64_t size)
std::pair< uint64_t, uint64_t > GetBlock()
XRootDStatus Initialize()
XRootDStatus GetChunk(XrdCl::PageInfo &ci)
static void DeleteChunk(PageInfo *&chunk)
const uint16_t stError
An error occurred that could potentially be retried.
const uint16_t errInternal
Internal error.
const uint16_t stOK
Everything went OK.
const uint64_t UtilityMsg
const uint16_t suContinue
const uint16_t errNoMoreReplicas
No more replicas to try.
uint32_t GetLength() const
Get the data length.