Binding media stream audio track to speech recognition [browser]

content/browser/speech/speech_recognition_audio_receiver.cc

Index: content/browser/speech/speech_recognition_audio_receiver.cc
diff --git a/content/browser/speech/speech_recognition_audio_receiver.cc b/content/browser/speech/speech_recognition_audio_receiver.cc
new file mode 100644
index 0000000000000000000000000000000000000000..8bd5ca828390e5ae529210fac2ac09c5d256a8c2
--- /dev/null
+++ b/content/browser/speech/speech_recognition_audio_receiver.cc
@@ -0,0 +1,226 @@
+// Copyright 2014 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 "content/browser/speech/speech_recognition_audio_receiver.h"
+
+#include "base/bind.h"
+#include "base/process/process_handle.h"
+#include "content/public/browser/browser_thread.h"
+#include "media/audio/audio_parameters.h"
+#include "media/base/audio_bus.h"
+
+namespace content {
+
+// static
+scoped_ptr<SpeechRecognitionAudioReceiver>
+  SpeechRecognitionAudioReceiver::Create(
+    const media::AudioParameters& audio_params,
+    const base::ProcessHandle peer_process_handle,
+    media::AudioInputController::EventHandler* event_handler,
+    base::SingleThreadTaskRunner* task_runner,
+    base::SharedMemoryHandle* foreign_memory_handle,
+    base::SyncSocket::TransitDescriptor* foreign_socket_descriptor) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  DCHECK(event_handler);
+
+  scoped_ptr<SpeechRecognitionAudioReceiver> receiver(
+      new SpeechRecognitionAudioReceiver(audio_params,
+                                         event_handler,
+                                         task_runner));
+
+  if (!receiver->Initialize(peer_process_handle,
+                            foreign_memory_handle,
+                            foreign_socket_descriptor)) {
+    receiver.reset();
+  }
+
+  return receiver.Pass();
+}
+
+SpeechRecognitionAudioReceiver::SpeechRecognitionAudioReceiver(
+    const media::AudioParameters& audio_params,
+    media::AudioInputController::EventHandler* event_handler,
+    base::SingleThreadTaskRunner* task_runner)
+    : audio_params_(audio_params),
+      audio_chunk_duration_(base::TimeDelta::FromMilliseconds(
+          (audio_params_.frames_per_buffer() * 1000) /
+          audio_params_.sample_rate())),
+      task_runner_(task_runner),
+      event_handler_(event_handler),
+      dropped_packets_(0),
+      weak_ptr_factory_(this),
+      listening_(false) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  DCHECK(task_runner);
+}
+
+SpeechRecognitionAudioReceiver::~SpeechRecognitionAudioReceiver() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+
+  // Make sure no new tasks are getting posted to the |task_runner_| and that we
+  // close the socket since the receiver is going away.
+  Stop();

[tommi (sloooow) - chröme, 2014/10/16 21:20:23]

is this method only called internally? (if so, can we make it private?)

[burnik, 2014/10/17 10:48:01]

It's called by the client as well. Maybe just put a DCHECK(!IsListening()) here?

+}
+
+bool SpeechRecognitionAudioReceiver::Initialize(
+    const base::ProcessHandle peer_process_handle,
+    base::SharedMemoryHandle* foreign_memory_handle,
+    base::SyncSocket::TransitDescriptor* foreign_socket_descriptor) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  DCHECK(foreign_memory_handle);
+  DCHECK(foreign_socket_descriptor);
+  DCHECK(!IsListening());
+
+  // All clean up on failure will be done by the destructor.
+  const size_t kSharedMemorySize = sizeof(media::AudioInputBufferParameters) +
+      media::AudioBus::CalculateMemorySize(audio_params_);
+  shared_memory_.reset(new base::SharedMemory());
+  if (!shared_memory_->CreateAndMapAnonymous(kSharedMemorySize)) {
+    DLOG(ERROR) << "Failed to create shared memory for the speech recognizer";
+    return false;
+  }
+  memset(shared_memory_->memory(), 0, kSharedMemorySize);
+
+  if (!shared_memory_->ShareToProcess(base::GetCurrentProcessHandle(),
+                                      foreign_memory_handle)) {
+    DLOG(ERROR) << "Failed to map and share the shared memory";
+    return false;
+  }
+
+  foreign_socket_.reset(new base::CancelableSyncSocket());
+  receiving_socket_.reset(new base::CancelableSyncSocket());
+  if (!base::SyncSocket::CreatePair(receiving_socket_.get(),
+                                    foreign_socket_.get())) {
+    DLOG(ERROR) << "Cannot create SyncSocket pair";
+    return false;
+  }
+
+  if (!foreign_socket_->PrepareTransitDescriptor(peer_process_handle,
+                                                 foreign_socket_descriptor)) {
+    DLOG(ERROR) << "Cannot prepare syncsocket transit descriptor";
+    return false;
+  }
+
+  media::AudioInputBuffer* audio_buffer =
+      static_cast<media::AudioInputBuffer*>(shared_memory_->memory());
+  audio_bus_ = media::AudioBus::WrapMemory(audio_params_, audio_buffer->audio);
+  audio_bus_->Zero();
+
+  // Use |buffer->params.size| from shared memory to synchronize by counting
+  // audio buffers when receiving from the renderer.
+  audio_buffer->params.size = 0;
+
+  thread_checker_.DetachFromThread();
+
+  return true;
+}
+
+void SpeechRecognitionAudioReceiver::Listen() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  if (IsListening())
+    return;
+
+  // It will take at least 2 * |audio_chunk_duration_| ms for the producer to
+  // send the first packet, therefore we delay the task. We also allow a 100%
+  // time overhead on the socket receive timeout because the first packet
+  // usually arrives late (depending on the IPC and the renderer).
+  task_runner_->PostDelayedTask(
+      FROM_HERE,
+      base::Bind(&SpeechRecognitionAudioReceiver::Receive,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 audio_chunk_duration_),
+      2 * audio_chunk_duration_);
+
+  SetListening(true);
+}
+
+void SpeechRecognitionAudioReceiver::Stop() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  if (!IsListening())
+    return;
+
+  SetListening(false);
+
+  // We close the socket so the task runner thread doesn't block on it.
+  receiving_socket_->Shutdown();
+}
+
+void SpeechRecognitionAudioReceiver::Receive(
+    const base::TimeDelta& receive_timeout) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK_LT(receive_timeout.InMilliseconds(), 1000);
+
+  // Maximum of consecutive dropped packets. When reached we trigger an error
+  // and stop reposting the task for |Receive()|.
+  static const uint32 kMaxDroppedPackets = 2;
+  static const size_t kSize = sizeof(media::AudioInputBufferParameters().size);
+
+  // Receive with timeout not to block indefinitely. Ideally we block here for
+  // no time at all for all packets after the first one.
+  size_t bytes_received = receiving_socket_->ReceiveWithTimeout(
+      &(GetAudioBuffer()->params.size), kSize, receive_timeout);
+  DCHECK_EQ(0U, bytes_received % kSize);
+
+  // Register time to account for any delay before we repost the task.
+  const base::TimeTicks last_receive_time = base::TimeTicks::Now();
+
+  if (bytes_received == kSize) {
+    // Any previous issues with the socket are ignored.
+    dropped_packets_ = 0;
+
+    // Execute the client callback on this |task_runner_|'s thread.
+    event_handler_->OnData(NULL, audio_bus_.get());
+
+    // Notify host via shared memory that the buffer was consumed.
+    GetAudioBuffer()->params.size++;
+  } else {
+    // If Stop() was called we stop reposting the task.
+    if (!IsListening())
+      return;
+
+    // Socket issue: either a timeout occurred or socket is closed.
+    if (++dropped_packets_ >= kMaxDroppedPackets) {
+      DLOG(ERROR) << "Dropped max packets";
+      receiving_socket_->Shutdown();
+
+      // We have passed the limit of consecutive dropped packets so we trigger
+      // the error event.
+      event_handler_->OnError(
+          NULL, media::AudioInputController::UNKNOWN_ERROR);
+
+      return;
+    }
+  }
+
+  const base::TimeDelta time_delta = base::TimeTicks::Now() - last_receive_time;
+  DCHECK_LE(time_delta.InMilliseconds(),
+            audio_chunk_duration_.InMilliseconds());
+
+  // Repost the task with allowing 50% delay on the socket receive.
+  base::MessageLoopProxy::current()->PostDelayedTask(
+      FROM_HERE,
+      base::Bind(&SpeechRecognitionAudioReceiver::Receive,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 audio_chunk_duration_ / 2),
+      audio_chunk_duration_ - time_delta);
+}
+
+media::AudioInputBuffer* SpeechRecognitionAudioReceiver::GetAudioBuffer() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  DCHECK_GE(shared_memory_->mapped_size(), shared_memory_->requested_size());
+  return static_cast<media::AudioInputBuffer*>(shared_memory_->memory());
+}
+
+bool SpeechRecognitionAudioReceiver::IsListening() {
+  base::AutoLock lock(listening_lock_);
+  return listening_;
+}
+
+void SpeechRecognitionAudioReceiver::SetListening(const bool listening) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  base::AutoLock lock(listening_lock_);
+  listening_ = listening;
+}
+
+}  // namespace content