Fix AudioPump to pause the stream when the network is congested.

remoting/host/audio_pump.cc

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "remoting/host/audio_pump.h"
 
 #include "base/bind.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/single_thread_task_runner.h"
 #include "base/thread_task_runner_handle.h"
 #include "remoting/codec/audio_encoder.h"
 #include "remoting/host/audio_capturer.h"
 #include "remoting/proto/audio.pb.h"
 #include "remoting/protocol/audio_stub.h"
 
 namespace remoting {
 
+// Maximum number of bytes that can be stored in the send buffer before we start
+// dropping packets. 5kB is about 0.25s in a stream with 160kb/s bitrate.

[Wez, 2015/02/12 02:29:47]

Do we run our stream at a fixed bit-rate? If not then why is 160kb/s so special?
:)

Should we express this constant as a number-of-milliseconds and derived the
actual buffer size from that and the bit-rate (or a "usual" bit-rate value)?

[Sergey Ulanov, 2015/02/13 21:21:56]

Our opus encoder uses this bitrate. See
https://code.google.com/p/chromium/codesearch#chromium/src/remoting/codec/aud...
.
Added GetBitrate() in AudioEncoder interface.

+const int kMaxBufferedBytes = 5000;
+
 class AudioPump::Core : public base::NonThreadSafe {
  public:
   Core(base::WeakPtr<AudioPump> pump,
        scoped_ptr<AudioCapturer> audio_capturer,
        scoped_ptr<AudioEncoder> audio_encoder);
   ~Core();
 
   void Start();
   void Pause(bool pause);
 
+  void OnPacketSent(int size);
+
  private:
   void EncodeAudioPacket(scoped_ptr<AudioPacket> packet);
 
   base::WeakPtr<AudioPump> pump_;
-
   scoped_refptr<base::SingleThreadTaskRunner> caller_task_runner_;
-
   scoped_ptr<AudioCapturer> audio_capturer_;
-
   scoped_ptr<AudioEncoder> audio_encoder_;
 
   bool enabled_;
 
+  // Number of bytes in the queue that have been encoded but haven't been sent
+  // yet.
+  int bytes_pending_;
+
   DISALLOW_COPY_AND_ASSIGN(Core);
 };
 
-AudioPump::Core::Core(
-    base::WeakPtr<AudioPump> pump,
-    scoped_ptr<AudioCapturer> audio_capturer,
-    scoped_ptr<AudioEncoder> audio_encoder)
+AudioPump::Core::Core(base::WeakPtr<AudioPump> pump,
+                      scoped_ptr<AudioCapturer> audio_capturer,
+                      scoped_ptr<AudioEncoder> audio_encoder)
     : pump_(pump),
       caller_task_runner_(base::ThreadTaskRunnerHandle::Get()),
       audio_capturer_(audio_capturer.Pass()),
       audio_encoder_(audio_encoder.Pass()),
-      enabled_(true) {
+      enabled_(true),
+      bytes_pending_(0) {
   DetachFromThread();
 }
 
 AudioPump::Core::~Core() {
   DCHECK(CalledOnValidThread());
 }
 
 void AudioPump::Core::Start() {
   DCHECK(CalledOnValidThread());
 
   audio_capturer_->Start(
       base::Bind(&Core::EncodeAudioPacket, base::Unretained(this)));
 }
 
 void AudioPump::Core::Pause(bool pause) {
   DCHECK(CalledOnValidThread());
 
   enabled_ = !pause;
 }
 
+void AudioPump::Core::OnPacketSent(int size) {
+  DCHECK(CalledOnValidThread());
+
+  bytes_pending_-= size;
+  DCHECK_GE(bytes_pending_, 0);
+}
+
 void AudioPump::Core::EncodeAudioPacket(scoped_ptr<AudioPacket> packet) {
   DCHECK(CalledOnValidThread());
   DCHECK(packet);
 
-  if (!enabled_)
+  if (!enabled_ || bytes_pending_ > kMaxBufferedBytes)
     return;
 
+
   scoped_ptr<AudioPacket> encoded_packet =
       audio_encoder_->Encode(packet.Pass());
 
   // The audio encoder returns a null audio packet if there's no audio to send.
   if (!encoded_packet)
     return;
 
-  caller_task_runner_->PostTask(FROM_HERE,
-                                 base::Bind(&AudioPump::SendAudioPacket, pump_,
-                                            base::Passed(&encoded_packet)));
+  int packet_size = encoded_packet->ByteSize();
+  bytes_pending_ += packet_size;
+
+  caller_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&AudioPump::SendAudioPacket, pump_,
+                            base::Passed(&encoded_packet), packet_size));
 }
 
 AudioPump::AudioPump(
     scoped_refptr<base::SingleThreadTaskRunner> audio_task_runner,
     scoped_ptr<AudioCapturer> audio_capturer,
     scoped_ptr<AudioEncoder> audio_encoder,
     protocol::AudioStub* audio_stub)
     : audio_task_runner_(audio_task_runner),
       audio_stub_(audio_stub),
       weak_factory_(this) {
@@ skipping 13 matching lines @@
 }
 
 void AudioPump::Pause(bool pause) {
   DCHECK(CalledOnValidThread());
 
   audio_task_runner_->PostTask(
       FROM_HERE,
       base::Bind(&Core::Pause, base::Unretained(core_.get()), pause));
 }
 
-void AudioPump::SendAudioPacket(scoped_ptr<AudioPacket> packet) {
+void AudioPump::SendAudioPacket(scoped_ptr<AudioPacket> packet, int size) {
   DCHECK(CalledOnValidThread());
   DCHECK(packet);
 
-  audio_stub_->ProcessAudioPacket(packet.Pass(), base::Closure());
+  audio_stub_->ProcessAudioPacket(
+      packet.Pass(),
+      base::Bind(&AudioPump::OnPacketSent, weak_factory_.GetWeakPtr(), size));
+}
+
+void AudioPump::OnPacketSent(int size) {
+  audio_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&Core::OnPacketSent, base::Unretained(core_.get()), size));
 }
 
 }  // namespace remoting