OLD | NEW |
1 // Copyright (c) 2010 The Chromium OS Authors. All rights reserved. | 1 // Copyright (c) 2010 The Chromium OS Authors. All rights reserved. |
2 // Use of this source code is governed by a BSD-style license that can be | 2 // Use of this source code is governed by a BSD-style license that can be |
3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
4 | 4 |
5 #ifndef CHROMEOS_PLATFORM_UPDATE_ENGINE_MULTI_HTTP_FETCHER_H__ | 5 #ifndef CHROMEOS_PLATFORM_UPDATE_ENGINE_MULTI_HTTP_FETCHER_H__ |
6 #define CHROMEOS_PLATFORM_UPDATE_ENGINE_MULTI_HTTP_FETCHER_H__ | 6 #define CHROMEOS_PLATFORM_UPDATE_ENGINE_MULTI_HTTP_FETCHER_H__ |
7 | 7 |
8 #include <tr1/memory> | 8 #include <tr1/memory> |
9 #include <utility> | 9 #include <utility> |
10 #include <vector> | 10 #include <vector> |
11 | 11 |
12 #include "update_engine/http_fetcher.h" | 12 #include "update_engine/http_fetcher.h" |
| 13 #include "update_engine/utils.h" |
13 | 14 |
14 // This class is a simple wrapper around an HttpFetcher. The client | 15 // This class is a simple wrapper around an HttpFetcher. The client |
15 // specifies a vector of byte ranges. MultiHttpFetcher will fetch bytes | 16 // specifies a vector of byte ranges. MultiHttpFetcher will fetch bytes |
16 // from those offsets. Pass -1 as a length to specify unlimited length. | 17 // from those offsets. Pass -1 as a length to specify unlimited length. |
17 // It really only would make sense for the last range specified to have | 18 // It really only would make sense for the last range specified to have |
18 // unlimited length. | 19 // unlimited length. |
19 | 20 |
20 namespace chromeos_update_engine { | 21 namespace chromeos_update_engine { |
21 | 22 |
22 template<typename BaseHttpFetcher> | 23 template<typename BaseHttpFetcher> |
23 class MultiHttpFetcher : public HttpFetcher, public HttpFetcherDelegate { | 24 class MultiHttpFetcher : public HttpFetcher, public HttpFetcherDelegate { |
24 public: | 25 public: |
25 typedef std::vector<std::pair<off_t, off_t> > RangesVect; | 26 typedef std::vector<std::pair<off_t, off_t> > RangesVect; |
26 | 27 |
27 MultiHttpFetcher() | 28 MultiHttpFetcher() |
28 : sent_transfer_complete_(false), | 29 : sent_transfer_complete_(false), |
| 30 pending_next_fetcher_(false), |
29 current_index_(0), | 31 current_index_(0), |
30 bytes_received_this_fetcher_(0) {} | 32 bytes_received_this_fetcher_(0) {} |
31 ~MultiHttpFetcher() {} | 33 ~MultiHttpFetcher() {} |
32 | 34 |
33 void set_ranges(const RangesVect& ranges) { | 35 void set_ranges(const RangesVect& ranges) { |
34 ranges_ = ranges; | 36 ranges_ = ranges; |
35 fetchers_.resize(ranges_.size()); // Allocate the fetchers | 37 fetchers_.resize(ranges_.size()); // Allocate the fetchers |
36 for (typename std::vector<std::tr1::shared_ptr<BaseHttpFetcher> | 38 for (typename std::vector<std::tr1::shared_ptr<BaseHttpFetcher> |
37 >::iterator it = fetchers_.begin(), e = fetchers_.end(); | 39 >::iterator it = fetchers_.begin(), e = fetchers_.end(); |
38 it != e; ++it) { | 40 it != e; ++it) { |
(...skipping 10 matching lines...) Expand all Loading... |
49 if (ranges_.empty()) { | 51 if (ranges_.empty()) { |
50 if (delegate_) | 52 if (delegate_) |
51 delegate_->TransferComplete(this, true); | 53 delegate_->TransferComplete(this, true); |
52 return; | 54 return; |
53 } | 55 } |
54 current_index_ = 0; | 56 current_index_ = 0; |
55 LOG(INFO) << "starting first transfer"; | 57 LOG(INFO) << "starting first transfer"; |
56 StartTransfer(); | 58 StartTransfer(); |
57 } | 59 } |
58 | 60 |
59 void TerminateTransfer() { | 61 void TransferTerminated(HttpFetcher* fetcher) { |
60 if (current_index_ < fetchers_.size()) | 62 LOG(INFO) << "Received transfer terminated."; |
61 fetchers_[current_index_]->TerminateTransfer(); | 63 if (pending_next_fetcher_) { |
| 64 pending_next_fetcher_ = false; |
| 65 if (++current_index_ >= ranges_.size()) { |
| 66 SendTransferComplete(fetcher, true); |
| 67 } else { |
| 68 StartTransfer(); |
| 69 } |
| 70 return; |
| 71 } |
62 current_index_ = ranges_.size(); | 72 current_index_ = ranges_.size(); |
63 sent_transfer_complete_ = true; // a fib | 73 sent_transfer_complete_ = true; // a fib |
| 74 if (delegate_) { |
| 75 // Note that after the callback returns this object may be destroyed. |
| 76 delegate_->TransferTerminated(this); |
| 77 } |
| 78 } |
| 79 |
| 80 void TerminateTransfer() { |
| 81 // If the current fetcher is already being terminated, just wait for its |
| 82 // TransferTerminated callback. |
| 83 if (pending_next_fetcher_) { |
| 84 pending_next_fetcher_ = false; |
| 85 return; |
| 86 } |
| 87 // If there's a current active fetcher terminate it and wait for its |
| 88 // TransferTerminated callback. |
| 89 if (current_index_ < fetchers_.size()) { |
| 90 fetchers_[current_index_]->TerminateTransfer(); |
| 91 return; |
| 92 } |
| 93 // Transfer is terminated before it got started and before any ranges were |
| 94 // added. |
| 95 TransferTerminated(this); |
64 } | 96 } |
65 | 97 |
66 void Pause() { | 98 void Pause() { |
67 if (current_index_ < fetchers_.size()) | 99 if (current_index_ < fetchers_.size()) |
68 fetchers_[current_index_]->Pause(); | 100 fetchers_[current_index_]->Pause(); |
69 } | 101 } |
70 | 102 |
71 void Unpause() { | 103 void Unpause() { |
72 if (current_index_ < fetchers_.size()) | 104 if (current_index_ < fetchers_.size()) |
73 fetchers_[current_index_]->Unpause(); | 105 fetchers_[current_index_]->Unpause(); |
(...skipping 46 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
120 } | 152 } |
121 LOG(INFO) << "Starting a transfer @" << ranges_[current_index_].first << "(" | 153 LOG(INFO) << "Starting a transfer @" << ranges_[current_index_].first << "(" |
122 << ranges_[current_index_].second << ")"; | 154 << ranges_[current_index_].second << ")"; |
123 bytes_received_this_fetcher_ = 0; | 155 bytes_received_this_fetcher_ = 0; |
124 fetchers_[current_index_]->SetOffset(ranges_[current_index_].first); | 156 fetchers_[current_index_]->SetOffset(ranges_[current_index_].first); |
125 if (delegate_) | 157 if (delegate_) |
126 delegate_->SeekToOffset(ranges_[current_index_].first); | 158 delegate_->SeekToOffset(ranges_[current_index_].first); |
127 fetchers_[current_index_]->BeginTransfer(url_); | 159 fetchers_[current_index_]->BeginTransfer(url_); |
128 } | 160 } |
129 | 161 |
130 void ReceivedBytes(HttpFetcher* fetcher, | 162 void ReceivedBytes(HttpFetcher* fetcher, const char* bytes, int length) { |
131 const char* bytes, | 163 TEST_AND_RETURN(current_index_ < ranges_.size()); |
132 int length) { | 164 TEST_AND_RETURN(fetcher == fetchers_[current_index_].get()); |
133 if (current_index_ >= ranges_.size()) | 165 TEST_AND_RETURN(!pending_next_fetcher_); |
134 return; | |
135 if (fetcher != fetchers_[current_index_].get()) { | |
136 LOG(WARNING) << "Received bytes from invalid fetcher"; | |
137 return; | |
138 } | |
139 off_t next_size = length; | 166 off_t next_size = length; |
140 if (ranges_[current_index_].second >= 0) { | 167 if (ranges_[current_index_].second >= 0) { |
141 next_size = std::min(next_size, | 168 next_size = std::min(next_size, |
142 ranges_[current_index_].second - | 169 ranges_[current_index_].second - |
143 bytes_received_this_fetcher_); | 170 bytes_received_this_fetcher_); |
144 } | 171 } |
145 LOG_IF(WARNING, next_size <= 0) << "Asked to write length <= 0"; | 172 LOG_IF(WARNING, next_size <= 0) << "Asked to write length <= 0"; |
146 if (delegate_) { | 173 if (delegate_) { |
147 delegate_->ReceivedBytes(this, bytes, next_size); | 174 delegate_->ReceivedBytes(this, bytes, next_size); |
148 } | 175 } |
149 bytes_received_this_fetcher_ += length; | 176 bytes_received_this_fetcher_ += length; |
150 if (ranges_[current_index_].second >= 0 && | 177 if (ranges_[current_index_].second >= 0 && |
151 bytes_received_this_fetcher_ >= ranges_[current_index_].second) { | 178 bytes_received_this_fetcher_ >= ranges_[current_index_].second) { |
152 fetchers_[current_index_]->TerminateTransfer(); | 179 // Terminates the current fetcher. Waits for its TransferTerminated |
153 current_index_++; | 180 // callback before starting the next fetcher so that we don't end up |
154 if (current_index_ == ranges_.size()) { | 181 // signalling the delegate that the whole multi-transfer is complete |
155 SendTransferComplete(fetchers_[current_index_ - 1].get(), true); | 182 // before all fetchers are really done and cleaned up. |
156 } else { | 183 pending_next_fetcher_ = true; |
157 StartTransfer(); | 184 fetcher->TerminateTransfer(); |
158 } | |
159 } | 185 } |
160 } | 186 } |
161 | 187 |
162 void TransferComplete(HttpFetcher* fetcher, bool successful) { | 188 void TransferComplete(HttpFetcher* fetcher, bool successful) { |
163 LOG(INFO) << "Received transfer complete"; | 189 TEST_AND_RETURN(!pending_next_fetcher_); |
| 190 LOG(INFO) << "Received transfer complete."; |
164 if (current_index_ >= ranges_.size()) { | 191 if (current_index_ >= ranges_.size()) { |
165 SendTransferComplete(fetcher, true); | 192 SendTransferComplete(fetcher, true); |
166 return; | 193 return; |
167 } | 194 } |
168 | |
169 if (ranges_[current_index_].second < 0) { | 195 if (ranges_[current_index_].second < 0) { |
170 // We're done with the current operation | 196 // We're done with the current operation |
171 current_index_++; | 197 current_index_++; |
172 if (current_index_ >= ranges_.size() || !successful) { | 198 if (current_index_ >= ranges_.size() || !successful) { |
173 SendTransferComplete(fetcher, successful); | 199 SendTransferComplete(fetcher, successful); |
174 } else { | 200 } else { |
175 // Do the next transfer | 201 // Do the next transfer |
176 StartTransfer(); | 202 StartTransfer(); |
177 } | 203 } |
178 return; | 204 return; |
179 } | 205 } |
180 | |
181 if (bytes_received_this_fetcher_ < ranges_[current_index_].second) { | 206 if (bytes_received_this_fetcher_ < ranges_[current_index_].second) { |
182 LOG(WARNING) << "Received insufficient bytes from fetcher. " | 207 LOG(WARNING) << "Received insufficient bytes from fetcher. " |
183 << "Ending early"; | 208 << "Ending early"; |
184 SendTransferComplete(fetcher, false); | 209 SendTransferComplete(fetcher, false); |
185 return; | 210 return; |
186 } else { | |
187 LOG(INFO) << "Got spurious TransferComplete. Ingoring."; | |
188 } | 211 } |
| 212 LOG(INFO) << "Got spurious TransferComplete. Ignoring."; |
189 } | 213 } |
190 | 214 |
191 // If true, do not send any more data or TransferComplete to the delegate. | 215 // If true, do not send any more data or TransferComplete to the delegate. |
192 bool sent_transfer_complete_; | 216 bool sent_transfer_complete_; |
193 | 217 |
| 218 // If true, the next fetcher needs to be started when TransferTerminated is |
| 219 // received from the current fetcher. |
| 220 bool pending_next_fetcher_; |
| 221 |
194 RangesVect ranges_; | 222 RangesVect ranges_; |
195 std::vector<std::tr1::shared_ptr<BaseHttpFetcher> > fetchers_; | 223 std::vector<std::tr1::shared_ptr<BaseHttpFetcher> > fetchers_; |
196 | 224 |
197 RangesVect::size_type current_index_; // index into ranges_, fetchers_ | 225 RangesVect::size_type current_index_; // index into ranges_, fetchers_ |
198 off_t bytes_received_this_fetcher_; | 226 off_t bytes_received_this_fetcher_; |
199 | 227 |
200 private: | |
201 DISALLOW_COPY_AND_ASSIGN(MultiHttpFetcher); | 228 DISALLOW_COPY_AND_ASSIGN(MultiHttpFetcher); |
202 }; | 229 }; |
203 | 230 |
204 } // namespace chromeos_update_engine | 231 } // namespace chromeos_update_engine |
205 | 232 |
206 #endif // CHROMEOS_PLATFORM_UPDATE_ENGINE_MULTI_HTTP_FETCHER_H__ | 233 #endif // CHROMEOS_PLATFORM_UPDATE_ENGINE_MULTI_HTTP_FETCHER_H__ |
OLD | NEW |