XRootD
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"
28 #include "XrdCl/XrdClDefaultEnv.hh"
29 #include "XrdCl/XrdClConstants.hh"
30 
31 #include <algorithm>
32 
33 namespace XrdCl
34 {
35 
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 )
41 {
42  SetFileSize( fileSize );
43 }
44 
45 XCpCtx::~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 
57 bool 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 
87 void XCpCtx::PutChunk( PageInfo* chunk )
88 {
89  pSink.Put( chunk );
90 }
91 
92 std::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 
104 void 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 
187 size_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()
Definition: XrdClXCpCtx.cc:172
bool GetNextUrl(std::string &url)
Definition: XrdClXCpCtx.cc:57
XCpSrc * WeakestLink(XCpSrc *exclude)
Definition: XrdClXCpCtx.cc:66
void PutChunk(PageInfo *chunk)
Definition: XrdClXCpCtx.cc:87
XCpCtx(const std::vector< std::string > &urls, uint64_t blockSize, uint8_t parallelSrc, uint64_t chunkSize, uint64_t parallelChunks, int64_t fileSize)
Definition: XrdClXCpCtx.cc:36
void SetFileSize(int64_t size)
Definition: XrdClXCpCtx.cc:104
std::pair< uint64_t, uint64_t > GetBlock()
Definition: XrdClXCpCtx.cc:92
XRootDStatus Initialize()
Definition: XrdClXCpCtx.cc:121
XRootDStatus GetChunk(XrdCl::PageInfo &ci)
Definition: XrdClXCpCtx.cc:140
static void DeleteChunk(PageInfo *&chunk)
Definition: XrdClXCpSrc.hh:126
uint64_t TransferRate()
Definition: XrdClXCpSrc.cc:584
const uint16_t suRetry
Definition: XrdClStatus.hh:40
const uint16_t stError
An error occurred that could potentially be retried.
Definition: XrdClStatus.hh:32
const uint16_t errInternal
Internal error.
Definition: XrdClStatus.hh:56
const uint16_t stOK
Everything went OK.
Definition: XrdClStatus.hh:31
const uint64_t UtilityMsg
const uint16_t suDone
Definition: XrdClStatus.hh:38
const uint16_t suContinue
Definition: XrdClStatus.hh:39
const uint16_t errNoMoreReplicas
No more replicas to try.
Definition: XrdClStatus.hh:65
uint32_t GetLength() const
Get the data length.