[Mac] Implement Ambient Light API
[chromium-blink-merge.git] / content / browser / byte_stream.cc
blob5c836bd6a0a87e5c1d580ea295f4c08c7adc37ea
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.
5 #include "content/browser/byte_stream.h"
7 #include <deque>
8 #include <set>
9 #include <utility>
11 #include "base/bind.h"
12 #include "base/location.h"
13 #include "base/memory/ref_counted.h"
14 #include "base/sequenced_task_runner.h"
16 namespace content {
17 namespace {
19 typedef std::deque<std::pair<scoped_refptr<net::IOBuffer>, size_t> >
20 ContentVector;
22 class ByteStreamReaderImpl;
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;
35 protected:
36 friend class base::RefCountedThreadSafe<LifetimeFlag>;
37 virtual ~LifetimeFlag() { }
39 private:
40 DISALLOW_COPY_AND_ASSIGN(LifetimeFlag);
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 ~ByteStreamWriterImpl() override;
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);
58 // Overridden from ByteStreamWriter.
59 bool Write(scoped_refptr<net::IOBuffer> buffer, size_t byte_count) override;
60 void Flush() override;
61 void Close(int status) override;
62 void RegisterCallback(const base::Closure& source_callback) override;
63 size_t GetTotalBufferedBytes() const override;
65 // PostTask target from |ByteStreamReaderImpl::MaybeUpdateInput|.
66 static void UpdateWindow(scoped_refptr<LifetimeFlag> lifetime_flag,
67 ByteStreamWriterImpl* target,
68 size_t bytes_consumed);
70 private:
71 // Called from UpdateWindow when object existence has been validated.
72 void UpdateWindowInternal(size_t bytes_consumed);
74 void PostToPeer(bool complete, int status);
76 const size_t total_buffer_size_;
78 // All data objects in this class are only valid to access on
79 // this task runner except as otherwise noted.
80 scoped_refptr<base::SequencedTaskRunner> my_task_runner_;
82 // True while this object is alive.
83 scoped_refptr<LifetimeFlag> my_lifetime_flag_;
85 base::Closure space_available_callback_;
86 ContentVector input_contents_;
87 size_t input_contents_size_;
89 // ** Peer information.
91 scoped_refptr<base::SequencedTaskRunner> peer_task_runner_;
93 // How much we've sent to the output that for flow control purposes we
94 // must assume hasn't been read yet.
95 size_t output_size_used_;
97 // Only valid to access on peer_task_runner_.
98 scoped_refptr<LifetimeFlag> peer_lifetime_flag_;
100 // Only valid to access on peer_task_runner_ if
101 // |*peer_lifetime_flag_ == true|
102 ByteStreamReaderImpl* peer_;
105 class ByteStreamReaderImpl : public ByteStreamReader {
106 public:
107 ByteStreamReaderImpl(scoped_refptr<base::SequencedTaskRunner> task_runner,
108 scoped_refptr<LifetimeFlag> lifetime_flag,
109 size_t buffer_size);
110 ~ByteStreamReaderImpl() override;
112 // Must be called before any operations are performed.
113 void SetPeer(ByteStreamWriterImpl* peer,
114 scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
115 scoped_refptr<LifetimeFlag> peer_lifetime_flag);
117 // Overridden from ByteStreamReader.
118 StreamState Read(scoped_refptr<net::IOBuffer>* data, size_t* length) override;
119 int GetStatus() const override;
120 void RegisterCallback(const base::Closure& sink_callback) override;
122 // PostTask target from |ByteStreamWriterImpl::Write| and
123 // |ByteStreamWriterImpl::Close|.
124 // Receive data from our peer.
125 // static because it may be called after the object it is targeting
126 // has been destroyed. It may not access |*target|
127 // if |*object_lifetime_flag| is false.
128 static void TransferData(
129 scoped_refptr<LifetimeFlag> object_lifetime_flag,
130 ByteStreamReaderImpl* target,
131 scoped_ptr<ContentVector> transfer_buffer,
132 size_t transfer_buffer_bytes,
133 bool source_complete,
134 int status);
136 private:
137 // Called from TransferData once object existence has been validated.
138 void TransferDataInternal(
139 scoped_ptr<ContentVector> transfer_buffer,
140 size_t transfer_buffer_bytes,
141 bool source_complete,
142 int status);
144 void MaybeUpdateInput();
146 const size_t total_buffer_size_;
148 scoped_refptr<base::SequencedTaskRunner> my_task_runner_;
150 // True while this object is alive.
151 scoped_refptr<LifetimeFlag> my_lifetime_flag_;
153 ContentVector available_contents_;
155 bool received_status_;
156 int status_;
158 base::Closure data_available_callback_;
160 // Time of last point at which data in stream transitioned from full
161 // to non-full. Nulled when a callback is sent.
162 base::Time last_non_full_time_;
164 // ** Peer information
166 scoped_refptr<base::SequencedTaskRunner> peer_task_runner_;
168 // How much has been removed from this class that we haven't told
169 // the input about yet.
170 size_t unreported_consumed_bytes_;
172 // Only valid to access on peer_task_runner_.
173 scoped_refptr<LifetimeFlag> peer_lifetime_flag_;
175 // Only valid to access on peer_task_runner_ if
176 // |*peer_lifetime_flag_ == true|
177 ByteStreamWriterImpl* peer_;
180 ByteStreamWriterImpl::ByteStreamWriterImpl(
181 scoped_refptr<base::SequencedTaskRunner> task_runner,
182 scoped_refptr<LifetimeFlag> lifetime_flag,
183 size_t buffer_size)
184 : total_buffer_size_(buffer_size),
185 my_task_runner_(task_runner),
186 my_lifetime_flag_(lifetime_flag),
187 input_contents_size_(0),
188 output_size_used_(0),
189 peer_(NULL) {
190 DCHECK(my_lifetime_flag_.get());
191 my_lifetime_flag_->is_alive = true;
194 ByteStreamWriterImpl::~ByteStreamWriterImpl() {
195 // No RunsTasksOnCurrentThread() check to allow deleting a created writer
196 // before we start using it. Once started, should be deleted on the specified
197 // task runner.
198 my_lifetime_flag_->is_alive = false;
201 void ByteStreamWriterImpl::SetPeer(
202 ByteStreamReaderImpl* peer,
203 scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
204 scoped_refptr<LifetimeFlag> peer_lifetime_flag) {
205 peer_ = peer;
206 peer_task_runner_ = peer_task_runner;
207 peer_lifetime_flag_ = peer_lifetime_flag;
210 bool ByteStreamWriterImpl::Write(
211 scoped_refptr<net::IOBuffer> buffer, size_t byte_count) {
212 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
214 // Check overflow.
216 // TODO(tyoshino): Discuss with content/browser/download developer and if
217 // they're fine with, set smaller limit and make it configurable.
218 size_t space_limit = std::numeric_limits<size_t>::max() -
219 GetTotalBufferedBytes();
220 if (byte_count > space_limit) {
221 // TODO(tyoshino): Tell the user that Write() failed.
222 // Ignore input.
223 return false;
226 input_contents_.push_back(std::make_pair(buffer, byte_count));
227 input_contents_size_ += byte_count;
229 // Arbitrarily, we buffer to a third of the total size before sending.
230 if (input_contents_size_ > total_buffer_size_ / kFractionBufferBeforeSending)
231 PostToPeer(false, 0);
233 return GetTotalBufferedBytes() <= total_buffer_size_;
236 void ByteStreamWriterImpl::Flush() {
237 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
238 if (input_contents_size_ > 0)
239 PostToPeer(false, 0);
242 void ByteStreamWriterImpl::Close(int status) {
243 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
244 PostToPeer(true, status);
247 void ByteStreamWriterImpl::RegisterCallback(
248 const base::Closure& source_callback) {
249 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
250 space_available_callback_ = source_callback;
253 size_t ByteStreamWriterImpl::GetTotalBufferedBytes() const {
254 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
255 // This sum doesn't overflow since Write() fails if this sum is going to
256 // overflow.
257 return input_contents_size_ + output_size_used_;
260 // static
261 void ByteStreamWriterImpl::UpdateWindow(
262 scoped_refptr<LifetimeFlag> lifetime_flag, ByteStreamWriterImpl* target,
263 size_t bytes_consumed) {
264 // If the target object isn't alive anymore, we do nothing.
265 if (!lifetime_flag->is_alive) return;
267 target->UpdateWindowInternal(bytes_consumed);
270 void ByteStreamWriterImpl::UpdateWindowInternal(size_t bytes_consumed) {
271 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
273 bool was_above_limit = GetTotalBufferedBytes() > total_buffer_size_;
275 DCHECK_GE(output_size_used_, bytes_consumed);
276 output_size_used_ -= bytes_consumed;
278 // Callback if we were above the limit and we're now <= to it.
279 bool no_longer_above_limit = GetTotalBufferedBytes() <= total_buffer_size_;
281 if (no_longer_above_limit && was_above_limit &&
282 !space_available_callback_.is_null())
283 space_available_callback_.Run();
286 void ByteStreamWriterImpl::PostToPeer(bool complete, int status) {
287 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
288 // Valid contexts in which to call.
289 DCHECK(complete || 0 != input_contents_size_);
291 scoped_ptr<ContentVector> transfer_buffer;
292 size_t buffer_size = 0;
293 if (0 != input_contents_size_) {
294 transfer_buffer.reset(new ContentVector);
295 transfer_buffer->swap(input_contents_);
296 buffer_size = input_contents_size_;
297 output_size_used_ += input_contents_size_;
298 input_contents_size_ = 0;
300 peer_task_runner_->PostTask(
301 FROM_HERE, base::Bind(
302 &ByteStreamReaderImpl::TransferData,
303 peer_lifetime_flag_,
304 peer_,
305 base::Passed(&transfer_buffer),
306 buffer_size,
307 complete,
308 status));
311 ByteStreamReaderImpl::ByteStreamReaderImpl(
312 scoped_refptr<base::SequencedTaskRunner> task_runner,
313 scoped_refptr<LifetimeFlag> lifetime_flag,
314 size_t buffer_size)
315 : total_buffer_size_(buffer_size),
316 my_task_runner_(task_runner),
317 my_lifetime_flag_(lifetime_flag),
318 received_status_(false),
319 status_(0),
320 unreported_consumed_bytes_(0),
321 peer_(NULL) {
322 DCHECK(my_lifetime_flag_.get());
323 my_lifetime_flag_->is_alive = true;
326 ByteStreamReaderImpl::~ByteStreamReaderImpl() {
327 // No RunsTasksOnCurrentThread() check to allow deleting a created writer
328 // before we start using it. Once started, should be deleted on the specified
329 // task runner.
330 my_lifetime_flag_->is_alive = false;
333 void ByteStreamReaderImpl::SetPeer(
334 ByteStreamWriterImpl* peer,
335 scoped_refptr<base::SequencedTaskRunner> peer_task_runner,
336 scoped_refptr<LifetimeFlag> peer_lifetime_flag) {
337 peer_ = peer;
338 peer_task_runner_ = peer_task_runner;
339 peer_lifetime_flag_ = peer_lifetime_flag;
342 ByteStreamReaderImpl::StreamState
343 ByteStreamReaderImpl::Read(scoped_refptr<net::IOBuffer>* data,
344 size_t* length) {
345 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
347 if (available_contents_.size()) {
348 *data = available_contents_.front().first;
349 *length = available_contents_.front().second;
350 available_contents_.pop_front();
351 unreported_consumed_bytes_ += *length;
353 MaybeUpdateInput();
354 return STREAM_HAS_DATA;
356 if (received_status_) {
357 return STREAM_COMPLETE;
359 return STREAM_EMPTY;
362 int ByteStreamReaderImpl::GetStatus() const {
363 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
364 DCHECK(received_status_);
365 return status_;
368 void ByteStreamReaderImpl::RegisterCallback(
369 const base::Closure& sink_callback) {
370 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
372 data_available_callback_ = sink_callback;
375 // static
376 void ByteStreamReaderImpl::TransferData(
377 scoped_refptr<LifetimeFlag> object_lifetime_flag,
378 ByteStreamReaderImpl* target,
379 scoped_ptr<ContentVector> transfer_buffer,
380 size_t buffer_size,
381 bool source_complete,
382 int status) {
383 // If our target is no longer alive, do nothing.
384 if (!object_lifetime_flag->is_alive) return;
386 target->TransferDataInternal(
387 transfer_buffer.Pass(), buffer_size, source_complete, status);
390 void ByteStreamReaderImpl::TransferDataInternal(
391 scoped_ptr<ContentVector> transfer_buffer,
392 size_t buffer_size,
393 bool source_complete,
394 int status) {
395 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
397 bool was_empty = available_contents_.empty();
399 if (transfer_buffer) {
400 available_contents_.insert(available_contents_.end(),
401 transfer_buffer->begin(),
402 transfer_buffer->end());
405 if (source_complete) {
406 received_status_ = true;
407 status_ = status;
410 // Callback on transition from empty to non-empty, or
411 // source complete.
412 if (((was_empty && !available_contents_.empty()) ||
413 source_complete) &&
414 !data_available_callback_.is_null())
415 data_available_callback_.Run();
418 // Decide whether or not to send the input a window update.
419 // Currently we do that whenever we've got unreported consumption
420 // greater than 1/3 of total size.
421 void ByteStreamReaderImpl::MaybeUpdateInput() {
422 DCHECK(my_task_runner_->RunsTasksOnCurrentThread());
424 if (unreported_consumed_bytes_ <=
425 total_buffer_size_ / kFractionReadBeforeWindowUpdate)
426 return;
428 peer_task_runner_->PostTask(
429 FROM_HERE, base::Bind(
430 &ByteStreamWriterImpl::UpdateWindow,
431 peer_lifetime_flag_,
432 peer_,
433 unreported_consumed_bytes_));
434 unreported_consumed_bytes_ = 0;
437 } // namespace
439 const int ByteStreamWriter::kFractionBufferBeforeSending = 3;
440 const int ByteStreamReader::kFractionReadBeforeWindowUpdate = 3;
442 ByteStreamReader::~ByteStreamReader() { }
444 ByteStreamWriter::~ByteStreamWriter() { }
446 void CreateByteStream(
447 scoped_refptr<base::SequencedTaskRunner> input_task_runner,
448 scoped_refptr<base::SequencedTaskRunner> output_task_runner,
449 size_t buffer_size,
450 scoped_ptr<ByteStreamWriter>* input,
451 scoped_ptr<ByteStreamReader>* output) {
452 scoped_refptr<LifetimeFlag> input_flag(new LifetimeFlag());
453 scoped_refptr<LifetimeFlag> output_flag(new LifetimeFlag());
455 ByteStreamWriterImpl* in = new ByteStreamWriterImpl(
456 input_task_runner, input_flag, buffer_size);
457 ByteStreamReaderImpl* out = new ByteStreamReaderImpl(
458 output_task_runner, output_flag, buffer_size);
460 in->SetPeer(out, output_task_runner, output_flag);
461 out->SetPeer(in, input_task_runner, input_flag);
462 input->reset(in);
463 output->reset(out);
466 } // namespace content