XRootD
Loading...
Searching...
No Matches
XrdClXCpCtx.cc
Go to the documentation of this file.
1//------------------------------------------------------------------------------
2// Copyright (c) 2011-2017 by European Organization for Nuclear Research (CERN)
3// Author: Michal Simon <michal.simon@cern.ch>
4//------------------------------------------------------------------------------
5// This file is part of the XRootD software suite.
6//
7// XRootD is free software: you can redistribute it and/or modify
8// it under the terms of the GNU Lesser General Public License as published by
9// the Free Software Foundation, either version 3 of the License, or
10// (at your option) any later version.
11//
12// XRootD is distributed in the hope that it will be useful,
13// but WITHOUT ANY WARRANTY; without even the implied warranty of
14// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15// GNU General Public License for more details.
16//
17// You should have received a copy of the GNU Lesser General Public License
18// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
19//
20// In applying this licence, CERN does not waive the privileges and immunities
21// granted to it by virtue of its status as an Intergovernmental Organization
22// or submit itself to any jurisdiction.
23//------------------------------------------------------------------------------
24
25#include "XrdCl/XrdClXCpCtx.hh"
26#include "XrdCl/XrdClXCpSrc.hh"
27#include "XrdCl/XrdClLog.hh"
30
31#include <algorithm>
32
33namespace XrdCl
34{
35
36XCpCtx::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 )
41{
42 SetFileSize( fileSize );
43}
44
45XCpCtx::~XCpCtx()
46{
47 // at this point there's no concurrency
48 // this object dies as the last one
49 while( !pSink.IsEmpty() )
50 {
51 PageInfo *chunk = pSink.Get();
52 if( chunk )
53 XCpSrc::DeleteChunk( chunk );
54 }
55}
56
57bool XCpCtx::GetNextUrl( std::string & url )
58{
59 XrdSysMutexHelper lck( pMtx );
60 if( pUrls.empty() ) return false;
61 url = pUrls.front();
62 pUrls.pop();
63 return true;
64}
65
67{
68 uint64_t transferRate = -1; // set transferRate to max uint64 value
69 XCpSrc *ret = 0;
70
71 std::list<XCpSrc*>::iterator itr;
72 for( itr = pSources.begin() ; itr != pSources.end() ; ++itr )
73 {
74 XCpSrc *src = *itr;
75 if( src == exclude ) continue;
76 uint64_t tmp = src->TransferRate();
77 if( src->HasData() && tmp < transferRate )
78 {
79 ret = src;
80 transferRate = tmp;
81 }
82 }
83
84 return ret;
85}
86
88{
89 pSink.Put( chunk );
90}
91
92std::pair<uint64_t, uint64_t> XCpCtx::GetBlock()
93{
94 XrdSysMutexHelper lck( pMtx );
95
96 uint64_t blkSize = pBlockSize, offset = pOffset;
97 if( pOffset + blkSize > uint64_t( pFileSize ) )
98 blkSize = pFileSize - pOffset;
99 pOffset += blkSize;
100
101 return std::make_pair( offset, blkSize );
102}
103
104void XCpCtx::SetFileSize( int64_t size )
105{
106 XrdSysMutexHelper lck( pMtx );
107 if( pFileSize < 0 && size >= 0 )
108 {
109 XrdSysCondVarHelper lck( pFileSizeCV );
110 pFileSize = size;
111 pFileSizeCV.Broadcast();
112
113 if( pBlockSize > uint64_t( pFileSize ) / pParallelSrc )
114 pBlockSize = pFileSize / pParallelSrc;
115
116 if( pBlockSize < pChunkSize )
117 pBlockSize = pChunkSize;
118 }
119}
120
122{
123 for( uint8_t i = 0; i < pParallelSrc; ++i )
124 {
125 XCpSrc *src = new XCpSrc( pChunkSize, pParallelChunks, pFileSize, this );
126 pSources.push_back( src );
127 src->Start();
128 }
129
130 if( pSources.empty() )
131 {
132 Log *log = DefaultEnv::GetLog();
133 log->Error( UtilityMsg, "Failed to initialize (failed to create new threads)" );
134 return XRootDStatus( stError, errInternal, EAGAIN, "XCpCtx: failed to create new threads." );
135 }
136
137 return XRootDStatus();
138}
139
141{
142 // if we received all the data we are done here
143 if( pDataReceived == uint64_t( pFileSize ) )
144 {
145 XrdSysCondVarHelper lck( pDoneCV );
146 pDone = true;
147 pDoneCV.Broadcast();
148 return XRootDStatus( stOK, suDone );
149 }
150
151 // if we don't have active sources it means we failed
152 if( GetRunning() == 0 )
153 {
154 XrdSysCondVarHelper lck( pDoneCV );
155 pDone = true;
156 pDoneCV.Broadcast();
158 }
159
160 PageInfo *chunk = pSink.Get();
161 if( chunk )
162 {
163 pDataReceived += chunk->GetLength();
164 ci = std::move( *chunk );
165 delete chunk;
166 return XRootDStatus( stOK, suContinue );
167 }
168
169 return XRootDStatus( stOK, suRetry );
170}
171
173{
174 pDoneCV.Broadcast();
175}
176
178{
179 XrdSysCondVarHelper lck( pDoneCV );
180
181 if( !pDone )
182 pDoneCV.Wait( 60 );
183
184 return pDone;
185}
186
187size_t XCpCtx::GetRunning()
188{
189 // count active sources
190 size_t nbRunning = 0;
191 std::list<XCpSrc*>::iterator itr;
192 for( itr = pSources.begin() ; itr != pSources.end() ; ++ itr)
193 if( (*itr)->IsRunning() )
194 ++nbRunning;
195 return nbRunning;
196}
197
198
199} /* namespace XrdCl */
static Log * GetLog()
Get default log.
Handle diagnostics.
Definition XrdClLog.hh:101
void Error(uint64_t topic, const char *format,...)
Report an error.
Definition XrdClLog.cc:231
void NotifyIdleSrc()
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)
uint64_t TransferRate()
const uint16_t suRetry
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 suDone
const uint16_t suContinue
const uint16_t errNoMoreReplicas
No more replicas to try.
uint32_t GetLength() const
Get the data length.