src/content/browser/byte_stream.cc

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