| OLD | NEW |
|---|
| (Empty) |
| 1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. | |
| 2 // Use of this source code is governed by a BSD-style license that can be | |
| 3 // found in the LICENSE file. | |
| 4 | |
| 5 #include "content/browser/byte_stream.h" | |
| 6 | |
| 7 #include "base/bind.h" | |
| 8 #include "base/location.h" | |
| 9 #include "base/memory/ref_counted.h" | |
| 10 #include "base/memory/weak_ptr.h" | |
| 11 #include "base/sequenced_task_runner.h" | |
| 12 | |
| 13 namespace content { | |
| 14 namespace { | |
| 15 | |
| 16 typedef std::deque<std::pair<scoped_refptr<net::IOBuffer>, size_t> > | |
| 17 ContentVector; | |
| 18 | |
| 19 class ByteStreamReaderImpl; | |
| 20 | |
| 21 // A poor man's weak pointer; a RefCountedThreadSafe boolean that can be | |
| 22 // cleared in an object destructor and accessed to check for object | |
| 23 // existence. We can't use weak pointers because they're tightly tied to | |
| 24 // threads rather than task runners. | |
| 25 // TODO(rdsmith): A better solution would be extending weak pointers | |
| 26 // to support SequencedTaskRunners. | |
| 27 struct LifetimeFlag : public base::RefCountedThreadSafe<LifetimeFlag> { | |
| 28 public: | |
| 29 LifetimeFlag() : is_alive(true) { } | |
| 30 bool is_alive; | |
| 31 | |
| 32 protected: | |
| 33 friend class base::RefCountedThreadSafe<LifetimeFlag>; | |
| 34 virtual ~LifetimeFlag() { } | |
| 35 | |
| 36 private: | |
| 37 DISALLOW_COPY_AND_ASSIGN(LifetimeFlag); | |
| 38 }; | |
| 39 | |
| 40 // For both ByteStreamWriterImpl and ByteStreamReaderImpl, Construction and | |
| 41 // SetPeer may happen anywhere; all other operations on each class must | |
| 42 // happen in the context of their SequencedTaskRunner. | |
| 43 class ByteStreamWriterImpl : public ByteStreamWriter { | |
| 44 public: | |
| 45 ByteStreamWriterImpl(scoped_refptr<base::SequencedTaskRunner> task_runner, | |
| 46 scoped_refptr<LifetimeFlag> lifetime_flag, | |
| 47 size_t buffer_size); | |
| 48 virtual ~ByteStreamWriterImpl(); | |
| 49 | |
| 50 // Must be called before any operations are performed. | |
| 51 void SetPeer(ByteStreamReaderImpl* peer, | |
| 52 scoped_refptr<base::SequencedTaskRunner> peer_task_runner, | |
| 53 scoped_refptr<LifetimeFlag> peer_lifetime_flag); | |
| 54 | |
| 55 // Overridden from ByteStreamWriter. | |
| 56 virtual bool Write(scoped_refptr<net::IOBuffer> buffer, | |
| 57 size_t byte_count) OVERRIDE; | |
| 58 virtual void Close(DownloadInterruptReason status) OVERRIDE; | |
| 59 virtual void RegisterCallback(const base::Closure& source_callback) OVERRIDE; | |
| 60 | |
| 61 // PostTask target from |ByteStreamReaderImpl::MaybeUpdateInput|. | |
| 62 static void UpdateWindow(scoped_refptr<LifetimeFlag> lifetime_flag, | |
| 63 ByteStreamWriterImpl* target, | |
| 64 size_t bytes_consumed); | |
| 65 | |
| 66 private: | |
| 67 // Called from UpdateWindow when object existence has been validated. | |
| 68 void UpdateWindowInternal(size_t bytes_consumed); | |
| 69 | |
| 70 void PostToPeer(bool complete, DownloadInterruptReason status); | |
| 71 | |
| 72 const size_t total_buffer_size_; | |
| 73 | |
| 74 // All data objects in this class are only valid to access on | |
| 75 // this task runner except as otherwise noted. | |
| 76 scoped_refptr<base::SequencedTaskRunner> my_task_runner_; | |
| 77 | |
| 78 // True while this object is alive. | |
| 79 scoped_refptr<LifetimeFlag> my_lifetime_flag_; | |
| 80 | |
| 81 base::Closure space_available_callback_; | |
| 82 ContentVector input_contents_; | |
| 83 size_t input_contents_size_; | |
| 84 | |
| 85 // ** Peer information. | |
| 86 | |
| 87 scoped_refptr<base::SequencedTaskRunner> peer_task_runner_; | |
| 88 | |
| 89 // How much we've sent to the output that for flow control purposes we | |
| 90 // must assume hasn't been read yet. | |
| 91 size_t output_size_used_; | |
| 92 | |
| 93 // Only valid to access on peer_task_runner_. | |
| 94 scoped_refptr<LifetimeFlag> peer_lifetime_flag_; | |
| 95 | |
| 96 // Only valid to access on peer_task_runner_ if | |
| 97 // |*peer_lifetime_flag_ == true| | |
| 98 ByteStreamReaderImpl* peer_; | |
| 99 }; | |
| 100 | |
| 101 class ByteStreamReaderImpl : public ByteStreamReader { | |
| 102 public: | |
| 103 ByteStreamReaderImpl(scoped_refptr<base::SequencedTaskRunner> task_runner, | |
| 104 scoped_refptr<LifetimeFlag> lifetime_flag, | |
| 105 size_t buffer_size); | |
| 106 virtual ~ByteStreamReaderImpl(); | |
| 107 | |
| 108 // Must be called before any operations are performed. | |
| 109 void SetPeer(ByteStreamWriterImpl* peer, | |
| 110 scoped_refptr<base::SequencedTaskRunner> peer_task_runner, | |
| 111 scoped_refptr<LifetimeFlag> peer_lifetime_flag); | |
| 112 | |
| 113 // Overridden from ByteStreamReader. | |
| 114 virtual StreamState Read(scoped_refptr<net::IOBuffer>* data, | |
| 115 size_t* length) OVERRIDE; | |
| 116 virtual DownloadInterruptReason GetStatus() const OVERRIDE; | |
| 117 virtual void RegisterCallback(const base::Closure& sink_callback) OVERRIDE; | |
| 118 | |
| 119 // PostTask target from |ByteStreamWriterImpl::MaybePostToPeer| and | |
| 120 // |ByteStreamWriterImpl::Close|. | |
| 121 // Receive data from our peer. | |
| 122 // static because it may be called after the object it is targeting | |
| 123 // has been destroyed. It may not access |*target| | |
| 124 // if |*object_lifetime_flag| is false. | |
| 125 static void TransferData( | |
| 126 scoped_refptr<LifetimeFlag> object_lifetime_flag, | |
| 127 ByteStreamReaderImpl* target, | |
| 128 scoped_ptr<ContentVector> transfer_buffer, | |
| 129 size_t transfer_buffer_bytes, | |
| 130 bool source_complete, | |
| 131 DownloadInterruptReason status); | |
| 132 | |
| 133 private: | |
| 134 // Called from TransferData once object existence has been validated. | |
| 135 void TransferDataInternal( | |
| 136 scoped_ptr<ContentVector> transfer_buffer, | |
| 137 size_t transfer_buffer_bytes, | |
| 138 bool source_complete, | |
| 139 DownloadInterruptReason status); | |
| 140 | |
| 141 void MaybeUpdateInput(); | |
| 142 | |
| 143 const size_t total_buffer_size_; | |
| 144 | |
| 145 scoped_refptr<base::SequencedTaskRunner> my_task_runner_; | |
| 146 | |
| 147 // True while this object is alive. | |
| 148 scoped_refptr<LifetimeFlag> my_lifetime_flag_; | |
| 149 | |
| 150 ContentVector available_contents_; | |
| 151 | |
| 152 bool received_status_; | |
| 153 DownloadInterruptReason status_; | |
| 154 | |
| 155 base::Closure data_available_callback_; | |
| 156 | |
| 157 // Time of last point at which data in stream transitioned from full | |
| 158 // to non-full. Nulled when a callback is sent. | |
| 159 base::Time last_non_full_time_; | |
| 160 | |
| 161 // ** Peer information | |
| 162 | |
| 163 scoped_refptr<base::SequencedTaskRunner> peer_task_runner_; | |
| 164 | |
| 165 // How much has been removed from this class that we haven't told | |
| 166 // the input about yet. | |
| 167 size_t unreported_consumed_bytes_; | |
| 168 | |
| 169 // Only valid to access on peer_task_runner_. | |
| 170 scoped_refptr<LifetimeFlag> peer_lifetime_flag_; | |
| 171 | |
| 172 // Only valid to access on peer_task_runner_ if | |
| 173 // |*peer_lifetime_flag_ == true| | |
| 174 ByteStreamWriterImpl* peer_; | |
| 175 }; | |
| 176 | |
| 177 ByteStreamWriterImpl::ByteStreamWriterImpl( | |
| 178 scoped_refptr<base::SequencedTaskRunner> task_runner, | |
| 179 scoped_refptr<LifetimeFlag> lifetime_flag, | |
| 180 size_t buffer_size) | |
| 181 : total_buffer_size_(buffer_size), | |
| 182 my_task_runner_(task_runner), | |
| 183 my_lifetime_flag_(lifetime_flag), | |
| 184 input_contents_size_(0), | |
| 185 output_size_used_(0), | |
| 186 peer_(NULL) { | |
| 187 DCHECK(my_lifetime_flag_.get()); | |
| 188 my_lifetime_flag_->is_alive = true; | |
| 189 } | |
| 190 | |
| 191 ByteStreamWriterImpl::~ByteStreamWriterImpl() { | |
| 192 my_lifetime_flag_->is_alive = false; | |
| 193 } | |
| 194 | |
| 195 void ByteStreamWriterImpl::SetPeer( | |
| 196 ByteStreamReaderImpl* peer, | |
| 197 scoped_refptr<base::SequencedTaskRunner> peer_task_runner, | |
| 198 scoped_refptr<LifetimeFlag> peer_lifetime_flag) { | |
| 199 peer_ = peer; | |
| 200 peer_task_runner_ = peer_task_runner; | |
| 201 peer_lifetime_flag_ = peer_lifetime_flag; | |
| 202 } | |
| 203 | |
| 204 bool ByteStreamWriterImpl::Write( | |
| 205 scoped_refptr<net::IOBuffer> buffer, size_t byte_count) { | |
| 206 DCHECK(my_task_runner_->RunsTasksOnCurrentThread()); | |
| 207 | |
| 208 input_contents_.push_back(std::make_pair(buffer, byte_count)); | |
| 209 input_contents_size_ += byte_count; | |
| 210 | |
| 211 // Arbitrarily, we buffer to a third of the total size before sending. | |
| 212 if (input_contents_size_ > total_buffer_size_ / kFractionBufferBeforeSending) | |
| 213 PostToPeer(false, DOWNLOAD_INTERRUPT_REASON_NONE); | |
| 214 | |
| 215 return (input_contents_size_ + output_size_used_ <= total_buffer_size_); | |
| 216 } | |
| 217 | |
| 218 void ByteStreamWriterImpl::Close( | |
| 219 DownloadInterruptReason status) { | |
| 220 DCHECK(my_task_runner_->RunsTasksOnCurrentThread()); | |
| 221 PostToPeer(true, status); | |
| 222 } | |
| 223 | |
| 224 void ByteStreamWriterImpl::RegisterCallback( | |
| 225 const base::Closure& source_callback) { | |
| 226 DCHECK(my_task_runner_->RunsTasksOnCurrentThread()); | |
| 227 space_available_callback_ = source_callback; | |
| 228 } | |
| 229 | |
| 230 // static | |
| 231 void ByteStreamWriterImpl::UpdateWindow( | |
| 232 scoped_refptr<LifetimeFlag> lifetime_flag, ByteStreamWriterImpl* target, | |
| 233 size_t bytes_consumed) { | |
| 234 // If the target object isn't alive anymore, we do nothing. | |
| 235 if (!lifetime_flag->is_alive) return; | |
| 236 | |
| 237 target->UpdateWindowInternal(bytes_consumed); | |
| 238 } | |
| 239 | |
| 240 void ByteStreamWriterImpl::UpdateWindowInternal(size_t bytes_consumed) { | |
| 241 DCHECK(my_task_runner_->RunsTasksOnCurrentThread()); | |
| 242 DCHECK_GE(output_size_used_, bytes_consumed); | |
| 243 output_size_used_ -= bytes_consumed; | |
| 244 | |
| 245 // Callback if we were above the limit and we're now <= to it. | |
| 246 size_t total_known_size_used = | |
| 247 input_contents_size_ + output_size_used_; | |
| 248 | |
| 249 if (total_known_size_used <= total_buffer_size_ && | |
| 250 (total_known_size_used + bytes_consumed > total_buffer_size_) && | |
| 251 !space_available_callback_.is_null()) | |
| 252 space_available_callback_.Run(); | |
| 253 } | |
| 254 | |
| 255 void ByteStreamWriterImpl::PostToPeer( | |
| 256 bool complete, DownloadInterruptReason status) { | |
| 257 DCHECK(my_task_runner_->RunsTasksOnCurrentThread()); | |
| 258 // Valid contexts in which to call. | |
| 259 DCHECK(complete || 0 != input_contents_size_); | |
| 260 | |
| 261 scoped_ptr<ContentVector> transfer_buffer(new ContentVector); | |
| 262 size_t buffer_size = 0; | |
| 263 if (0 != input_contents_size_) { | |
| 264 transfer_buffer.reset(new ContentVector); | |
| 265 transfer_buffer->swap(input_contents_); | |
| 266 buffer_size = input_contents_size_; | |
| 267 output_size_used_ += input_contents_size_; | |
| 268 input_contents_size_ = 0; | |
| 269 } | |
| 270 peer_task_runner_->PostTask( | |
| 271 FROM_HERE, base::Bind( | |
| 272 &ByteStreamReaderImpl::TransferData, | |
| 273 peer_lifetime_flag_, | |
| 274 peer_, | |
| 275 base::Passed(&transfer_buffer), | |
| 276 buffer_size, | |
| 277 complete, | |
| 278 status)); | |
| 279 } | |
| 280 | |
| 281 ByteStreamReaderImpl::ByteStreamReaderImpl( | |
| 282 scoped_refptr<base::SequencedTaskRunner> task_runner, | |
| 283 scoped_refptr<LifetimeFlag> lifetime_flag, | |
| 284 size_t buffer_size) | |
| 285 : total_buffer_size_(buffer_size), | |
| 286 my_task_runner_(task_runner), | |
| 287 my_lifetime_flag_(lifetime_flag), | |
| 288 received_status_(false), | |
| 289 status_(DOWNLOAD_INTERRUPT_REASON_NONE), | |
| 290 unreported_consumed_bytes_(0), | |
| 291 peer_(NULL) { | |
| 292 DCHECK(my_lifetime_flag_.get()); | |
| 293 my_lifetime_flag_->is_alive = true; | |
| 294 } | |
| 295 | |
| 296 ByteStreamReaderImpl::~ByteStreamReaderImpl() { | |
| 297 my_lifetime_flag_->is_alive = false; | |
| 298 } | |
| 299 | |
| 300 void ByteStreamReaderImpl::SetPeer( | |
| 301 ByteStreamWriterImpl* peer, | |
| 302 scoped_refptr<base::SequencedTaskRunner> peer_task_runner, | |
| 303 scoped_refptr<LifetimeFlag> peer_lifetime_flag) { | |
| 304 peer_ = peer; | |
| 305 peer_task_runner_ = peer_task_runner; | |
| 306 peer_lifetime_flag_ = peer_lifetime_flag; | |
| 307 } | |
| 308 | |
| 309 ByteStreamReaderImpl::StreamState | |
| 310 ByteStreamReaderImpl::Read(scoped_refptr<net::IOBuffer>* data, | |
| 311 size_t* length) { | |
| 312 DCHECK(my_task_runner_->RunsTasksOnCurrentThread()); | |
| 313 | |
| 314 if (available_contents_.size()) { | |
| 315 *data = available_contents_.front().first; | |
| 316 *length = available_contents_.front().second; | |
| 317 available_contents_.pop_front(); | |
| 318 unreported_consumed_bytes_ += *length; | |
| 319 | |
| 320 MaybeUpdateInput(); | |
| 321 return STREAM_HAS_DATA; | |
| 322 } | |
| 323 if (received_status_) { | |
| 324 return STREAM_COMPLETE; | |
| 325 } | |
| 326 return STREAM_EMPTY; | |
| 327 } | |
| 328 | |
| 329 DownloadInterruptReason ByteStreamReaderImpl::GetStatus() const { | |
| 330 DCHECK(my_task_runner_->RunsTasksOnCurrentThread()); | |
| 331 DCHECK(received_status_); | |
| 332 return status_; | |
| 333 } | |
| 334 | |
| 335 void ByteStreamReaderImpl::RegisterCallback( | |
| 336 const base::Closure& sink_callback) { | |
| 337 DCHECK(my_task_runner_->RunsTasksOnCurrentThread()); | |
| 338 | |
| 339 data_available_callback_ = sink_callback; | |
| 340 } | |
| 341 | |
| 342 // static | |
| 343 void ByteStreamReaderImpl::TransferData( | |
| 344 scoped_refptr<LifetimeFlag> object_lifetime_flag, | |
| 345 ByteStreamReaderImpl* target, | |
| 346 scoped_ptr<ContentVector> transfer_buffer, | |
| 347 size_t buffer_size, | |
| 348 bool source_complete, | |
| 349 DownloadInterruptReason status) { | |
| 350 // If our target is no longer alive, do nothing. | |
| 351 if (!object_lifetime_flag->is_alive) return; | |
| 352 | |
| 353 target->TransferDataInternal( | |
| 354 transfer_buffer.Pass(), buffer_size, source_complete, status); | |
| 355 } | |
| 356 | |
| 357 void ByteStreamReaderImpl::TransferDataInternal( | |
| 358 scoped_ptr<ContentVector> transfer_buffer, | |
| 359 size_t buffer_size, | |
| 360 bool source_complete, | |
| 361 DownloadInterruptReason status) { | |
| 362 DCHECK(my_task_runner_->RunsTasksOnCurrentThread()); | |
| 363 | |
| 364 bool was_empty = available_contents_.empty(); | |
| 365 | |
| 366 if (transfer_buffer) { | |
| 367 available_contents_.insert(available_contents_.end(), | |
| 368 transfer_buffer->begin(), | |
| 369 transfer_buffer->end()); | |
| 370 } | |
| 371 | |
| 372 if (source_complete) { | |
| 373 received_status_ = true; | |
| 374 status_ = status; | |
| 375 } | |
| 376 | |
| 377 // Callback on transition from empty to non-empty, or | |
| 378 // source complete. | |
| 379 if (((was_empty && !available_contents_.empty()) || | |
| 380 source_complete) && | |
| 381 !data_available_callback_.is_null()) | |
| 382 data_available_callback_.Run(); | |
| 383 } | |
| 384 | |
| 385 // Decide whether or not to send the input a window update. | |
| 386 // Currently we do that whenever we've got unreported consumption | |
| 387 // greater than 1/3 of total size. | |
| 388 void ByteStreamReaderImpl::MaybeUpdateInput() { | |
| 389 DCHECK(my_task_runner_->RunsTasksOnCurrentThread()); | |
| 390 | |
| 391 if (unreported_consumed_bytes_ <= | |
| 392 total_buffer_size_ / kFractionReadBeforeWindowUpdate) | |
| 393 return; | |
| 394 | |
| 395 peer_task_runner_->PostTask( | |
| 396 FROM_HERE, base::Bind( | |
| 397 &ByteStreamWriterImpl::UpdateWindow, | |
| 398 peer_lifetime_flag_, | |
| 399 peer_, | |
| 400 unreported_consumed_bytes_)); | |
| 401 unreported_consumed_bytes_ = 0; | |
| 402 } | |
| 403 | |
| 404 } // namespace | |
| 405 | |
| 406 | |
| 407 const int ByteStreamWriter::kFractionBufferBeforeSending = 3; | |
| 408 const int ByteStreamReader::kFractionReadBeforeWindowUpdate = 3; | |
| 409 | |
| 410 ByteStreamReader::~ByteStreamReader() { } | |
| 411 | |
| 412 ByteStreamWriter::~ByteStreamWriter() { } | |
| 413 | |
| 414 void CreateByteStream( | |
| 415 scoped_refptr<base::SequencedTaskRunner> input_task_runner, | |
| 416 scoped_refptr<base::SequencedTaskRunner> output_task_runner, | |
| 417 size_t buffer_size, | |
| 418 scoped_ptr<ByteStreamWriter>* input, | |
| 419 scoped_ptr<ByteStreamReader>* output) { | |
| 420 scoped_refptr<LifetimeFlag> input_flag(new LifetimeFlag()); | |
| 421 scoped_refptr<LifetimeFlag> output_flag(new LifetimeFlag()); | |
| 422 | |
| 423 ByteStreamWriterImpl* in = new ByteStreamWriterImpl( | |
| 424 input_task_runner, input_flag, buffer_size); | |
| 425 ByteStreamReaderImpl* out = new ByteStreamReaderImpl( | |
| 426 output_task_runner, output_flag, buffer_size); | |
| 427 | |
| 428 in->SetPeer(out, output_task_runner, output_flag); | |
| 429 out->SetPeer(in, input_task_runner, input_flag); | |
| 430 input->reset(in); | |
| 431 output->reset(out); | |
| 432 } | |
| 433 | |
| 434 } // namespace content | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 18284005:
Make ByteStream independent from DownloadInterruptReason (Closed)
Base URL: svn://svn.chromium.org/chrome/trunk/src