1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
|
#include "platform/chunks_download_strategy.hpp"
#include "coding/file_writer.hpp"
#include "coding/file_reader.hpp"
#include "coding/varint.hpp"
#include "base/logging.hpp"
#include "std/algorithm.hpp"
namespace downloader
{
ChunksDownloadStrategy::ChunksDownloadStrategy(vector<string> const & urls)
{
// init servers list
for (size_t i = 0; i < urls.size(); ++i)
m_servers.push_back(ServerT(urls[i], SERVER_READY));
}
pair<ChunksDownloadStrategy::ChunkT *, int>
ChunksDownloadStrategy::GetChunk(RangeT const & range)
{
vector<ChunkT>::iterator i = lower_bound(m_chunks.begin(), m_chunks.end(), range.first, LessChunks());
if (i != m_chunks.end() && i->m_pos == range.first)
{
ASSERT_EQUAL ( (i+1)->m_pos, range.second + 1, () );
return pair<ChunkT *, int>(&(*i), distance(m_chunks.begin(), i));
}
else
{
LOG(LERROR, ("Downloader error. Invalid chunk range: ", range));
return pair<ChunkT *, int>(static_cast<ChunkT *>(0), -1);
}
}
void ChunksDownloadStrategy::InitChunks(int64_t fileSize, int64_t chunkSize, ChunkStatusT status)
{
m_chunks.reserve(fileSize / chunkSize + 2);
for (int64_t i = 0; i < fileSize; i += chunkSize)
m_chunks.push_back(ChunkT(i, status));
m_chunks.push_back(ChunkT(fileSize, CHUNK_AUX));
}
void ChunksDownloadStrategy::AddChunk(RangeT const & range, ChunkStatusT status)
{
ASSERT_LESS_OR_EQUAL ( range.first, range.second, () );
if (m_chunks.empty())
{
ASSERT_EQUAL ( range.first, 0, () );
m_chunks.push_back(ChunkT(range.first, status));
}
else
{
ASSERT_EQUAL ( m_chunks.back().m_pos, range.first, () );
m_chunks.back().m_status = status;
}
m_chunks.push_back(ChunkT(range.second + 1, CHUNK_AUX));
}
void ChunksDownloadStrategy::SaveChunks(int64_t fileSize, string const & fName)
{
if (!m_chunks.empty())
{
try
{
FileWriter w(fName);
WriteVarInt(w, fileSize);
w.Write(&m_chunks[0], sizeof(ChunkT) * m_chunks.size());
return;
}
catch (FileWriter::Exception const & e)
{
LOG(LWARNING, ("Can't save chunks to file", e.Msg()));
}
}
// Delete if no chunks or some error occured.
(void)FileWriter::DeleteFileX(fName);
}
int64_t ChunksDownloadStrategy::LoadOrInitChunks( string const & fName,
int64_t fileSize, int64_t chunkSize)
{
ASSERT ( fileSize > 0, () );
ASSERT ( chunkSize > 0, () );
try
{
FileReader r(fName);
ReaderSource<FileReader> src(r);
int64_t const readedSize = ReadVarInt<int64_t>(src);
if (readedSize == fileSize)
{
// Load chunks.
uint64_t const size = src.Size();
int const stSize = sizeof(ChunkT);
size_t const count = size / stSize;
ASSERT_EQUAL(size, stSize * count, ());
m_chunks.resize(count);
src.Read(&m_chunks[0], stSize * count);
// Reset status "downloading" to "free".
int64_t downloadedSize = 0;
for (size_t i = 0; i < count-1; ++i)
{
if (m_chunks[i].m_status != CHUNK_COMPLETE)
m_chunks[i].m_status = CHUNK_FREE;
else
downloadedSize += (m_chunks[i+1].m_pos - m_chunks[i].m_pos);
}
return downloadedSize;
}
}
catch (RootException const & e)
{
// Usually - file not exists or Reader::Exception.
LOG(LDEBUG, (e.Msg()));
}
InitChunks(fileSize, chunkSize);
return 0;
}
void ChunksDownloadStrategy::ChunkFinished(bool success, RangeT const & range)
{
pair<ChunkT *, int> res = GetChunk(range);
// find server which was downloading this chunk
if (res.first)
{
for (size_t s = 0; s < m_servers.size(); ++s)
{
if (m_servers[s].m_chunkIndex == res.second)
{
if (success)
{
// mark server as free and chunk as ready
m_servers[s].m_chunkIndex = SERVER_READY;
res.first->m_status = CHUNK_COMPLETE;
}
else
{
LOG(LINFO, ("Thread for url", m_servers[s].m_url,
"failed to download chunk number", m_servers[s].m_chunkIndex));
// remove failed server and mark chunk as free
m_servers.erase(m_servers.begin() + s);
res.first->m_status = CHUNK_FREE;
}
break;
}
}
}
}
ChunksDownloadStrategy::ResultT
ChunksDownloadStrategy::NextChunk(string & outUrl, RangeT & range)
{
// If no servers at all.
if (m_servers.empty())
return EDownloadFailed;
// Find first free server.
ServerT * server = 0;
for (size_t i = 0; i < m_servers.size(); ++i)
{
if (m_servers[i].m_chunkIndex == SERVER_READY)
{
server = &m_servers[i];
break;
}
}
if (server == 0)
return ENoFreeServers;
bool allChunksDownloaded = true;
// Find first free chunk.
for (size_t i = 0; i < m_chunks.size()-1; ++i)
{
switch (m_chunks[i].m_status)
{
case CHUNK_FREE:
server->m_chunkIndex = static_cast<int>(i);
outUrl = server->m_url;
range.first = m_chunks[i].m_pos;
range.second = m_chunks[i+1].m_pos - 1;
m_chunks[i].m_status = CHUNK_DOWNLOADING;
return ENextChunk;
case CHUNK_DOWNLOADING:
allChunksDownloaded = false;
break;
}
}
return (allChunksDownloaded ? EDownloadSucceeded : ENoFreeServers);
}
} // namespace downloader
|
