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 "base/threading/thread_restrictions.h"
+#include "content/public/browser/browser_thread.h"
+#include "media/audio/audio_parameters.h"
+#include "media/base/audio_bus.h"
+
+namespace content {
+
+// static
+SpeechRecognitionAudioReceiver* SpeechRecognitionAudioReceiver::Create(

[no longer working on chromium, 2014/10/15 12:14:28]

use scoped_ptr<SpeechRecognitionAudioReceiver>

[burnik, 2014/10/15 14:52:27]

Done.

+    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)) {
+    return receiver.release();
+  } else {
+    return NULL;
+  }
+}
+
+SpeechRecognitionAudioReceiver::SpeechRecognitionAudioReceiver(
+  const media::AudioParameters& audio_params,
+  media::AudioInputController::EventHandler* event_handler,
+  base::SingleThreadTaskRunner* task_runner) :
+    audio_params_(audio_params),
+    task_runner_(task_runner),
+    event_handler_(event_handler),
+    dropped_packets_(0),
+    weak_ptr_factory_(this),
+    listening_(false) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+}
+
+SpeechRecognitionAudioReceiver::~SpeechRecognitionAudioReceiver() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+
+  // Make sure we don't issue an OnError event since the client is
+  // deleting the object.
+  set_listening(false);
+
+  // We close the socket so the task runner thread doesn't block on it.
+  receiving_socket_->Shutdown();
+
+  // No audio thread indicates the client provided a TaskRunner.
+  if (!audio_thread_.get())
+    return;
+
+  // We need to stop the |audio_thread_| to properly clean up.
+  // We must allow the IO until the task in the thread's message loop completes.
+  // This should execute almost immediately in most of cases since we expect
+  // the sockets were closed and |Receive()| returned without reposting a task.
+  const bool previous_io_allowed = base::ThreadRestrictions::SetIOAllowed(true);
+  audio_thread_->Stop();
+  base::ThreadRestrictions::SetIOAllowed(previous_io_allowed);
+}
+
+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(!listening());
+
+  // 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;
+
+  audio_thread_checker_.DetachFromThread();
+
+  // Check if we need a new thread for the task runner.
+  if (!task_runner_) {
+    audio_thread_.reset(new base::Thread("AudioReceiverThread"));
+    audio_thread_->Start();
+    task_runner_ = audio_thread_->task_runner().get();
+  }
+
+  return true;
+}
+
+void SpeechRecognitionAudioReceiver::Listen() {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  if (listening())
+    return;
+
+  task_runner_->PostTask(FROM_HERE,
+                         base::Bind(&SpeechRecognitionAudioReceiver::Receive,
+                                    weak_ptr_factory_.GetWeakPtr()));
+  set_listening(true);
+}
+
+void SpeechRecognitionAudioReceiver::Receive() {
+  DCHECK(audio_thread_checker_.CalledOnValidThread());
+  static const int kAudioChunkLengthMs =
+      (audio_params_.frames_per_buffer() * 1000) / audio_params_.sample_rate();
+
+  // Allow near 100% overhead on receive timeout.
+  // This is well adjusted to the allowed latency on the renderer.
+  // See |kNumberOfBuffersInFifo| in SpeechRecognitionAudioSink implementation.
+  static const base::TimeDelta kReceiveTimeout =
+      base::TimeDelta::FromMilliseconds(kAudioChunkLengthMs * 2);
+  DCHECK_LT(kReceiveTimeout.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
+  // |kAudioChunkLengthMs| as long as the socket is open.
+  size_t bytes_received = receiving_socket_->ReceiveWithTimeout(

[no longer working on chromium, 2014/10/15 12:14:28]

I think the best solution here is to use a short timeout and PostDelayedTask.

For example:
// Use 3ms as timeout.
static const base::TimeDelta kReceiveTimeout =
base::TimeDelta::FromMilliseconds(3);
size_t bytes_received = receiving_socket_->ReceiveWithTimeout(..)
const base::TimeDelta next_receive_time;
if (bytes_received == kSize) {
  ..

  // Schedule next receive to be 90ms later.
  next_receive_time = base::TimeDelta::FromMilliseconds(90);
} else {
  ....
  // Schedule next receive to be 5ms later.
  next_receive_time = base::TimeDelta::FromMilliseconds(5);
}

PostDelayedTask(FROM_HERE, base::Bind(&SpeechRecognitionAudioReceiver::Receive,
weak_ptr_factory_.GetWeakPtr()), next_receive_time);

[burnik, 2014/10/15 14:52:27]

Yeah, I've tried that approach. The delay seems not to occur. I'll try it again
to see if it only applies to either browser thread or the audio thread. However,
it is a bit tricky.

+      &(GetAudioBuffer()->params.size), kSize, kReceiveTimeout);
+  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.
+    GetAudioBuffer()->params.size++;
+  } else {
+    // If destructor was called we stop posting tasks to the thread.
+    if (!listening())
+      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;
+    }
+  }
+
+  base::MessageLoopProxy::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&SpeechRecognitionAudioReceiver::Receive,
+                 weak_ptr_factory_.GetWeakPtr()));
+}
+
+media::AudioInputBuffer* SpeechRecognitionAudioReceiver::GetAudioBuffer() {
+  DCHECK(audio_thread_checker_.CalledOnValidThread());
+  DCHECK_GE(shared_memory_->mapped_size(), shared_memory_->requested_size());
+  return static_cast<media::AudioInputBuffer*>(shared_memory_->memory());
+}
+
+bool SpeechRecognitionAudioReceiver::listening() {
+  base::AutoLock lock(listening_lock_);
+  return listening_;
+}
+
+void SpeechRecognitionAudioReceiver::set_listening(const bool listening) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
+  base::AutoLock lock(listening_lock_);
+  listening_ = listening;
+}
+
+}  // namespace content