Binding media stream audio track to speech recognition [browser]

content/browser/speech/speech_recognition_audio_receiver.cc

+// 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),
+      listening_(false) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  DCHECK(task_runner);
+}
+
+SpeechRecognitionAudioReceiver::~SpeechRecognitionAudioReceiver() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  // Make sure Stop() was called before we destroy the object.
+  DCHECK(!IsListening());
+}
+
+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 |audio_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::StartListening() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  if (IsListening())
+    return;
+  start_event_.reset(new base::WaitableEvent(false, false));
+  task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&SpeechRecognitionAudioReceiver::Receive,
+                 base::Unretained(this)));
+  start_event_->Wait();
+  start_event_.reset();
+  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() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // We use the audio buffer wrapping the shared memory to read audio data and
+  // write the buffer index via |audio_buffer->params.size| to synchronize.
+  media::AudioInputBuffer* audio_buffer = GetAudioBuffer();
+
+  // Maximum of consecutive dropped packets. When reached we trigger an error
+  // and stop the thread loop.
+  static const uint32 kMaxDroppedPackets = 2;
+  static const size_t kSize = sizeof(audio_buffer->params.size);
+  uint32 dropped_packets = 0U;
+
+  // The first packet will arrive late, so we use a longer first timeout.
+  base::TimeDelta receive_timeout = base::TimeDelta::FromMilliseconds(900);
+
+  // Signal the event handler that listening has started.
+  event_handler_->OnRecording(NULL);
+
+  // Signal the start event for thread synchronization.
+  DCHECK(start_event_.get());
+  start_event_->Signal();
+
+  while (true) {
+    size_t bytes_received = receiving_socket_->ReceiveWithTimeout(
+        &(audio_buffer->params.size), kSize, receive_timeout);
+    DCHECK_EQ(0U, bytes_received % kSize);
+
+    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.
+      ++(audio_buffer->params.size);
+    } else {
+      // If Stop() was called we stop the loop.
+      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;
+      }
+    }
+
+    // All packets after first one use a shorter timeout.
+    receive_timeout = audio_chunk_duration_ * 2;
+  }
+}
+
+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) {
+  base::AutoLock lock(listening_lock_);
+  listening_ = listening;
+}
+
+}  // namespace content