Bug 1940906 - Introduce SpeechRecognitionBackend - Part 1 - Static methods. r=nika,pehrsons

This patch starts filling in the blanks in SpeechRecognitionBackend. This
object sits behind the SpeechRecogntion DOM object, and implements some light
audio processing, and IPC communication.

Audio processing is left for the next patch. Beyond the static methods
(`available` and `install`, which call the IPDL methods of the same name),
this patch also implements the threading architecture, the IPC-thread
lifecycle, and the transient-vs-long-running session management described
below.

More importantly this patch implement a threading architecture that works for
real-time audio, and DOM API, static and instance methods.

The constraints are the following:
- We're going to send method calls (available, install, start, stop, etc.) to
the HWInference process, those come from the main thread (Web Speech API)
- We're also going to need to send real-time audio to the same process (with a
graduingly lower and lower packet size, to reduce end-to-end latency). This come
from a hard-real-time thread (from our MediaTrackGraph)
- Our IPC system requires all method calls to be on a single thread
- The real-time audio thread can't do any system calls or unbounded operations
(can't do IPC, locking, allocations, etc.)
- We really don't want to bounce the real-time audio via the main thread for IPC
or make it more busy if we can avoid it (it's already a nightmare in a typical
media processing web app)
- We can't affort scheduling hazards associated with using a shared thread (e.g.
the shared background thread), because we want to have low and constant latency
- There are both static and instance methods to implement

The design is therefore the following:
- A dedicated SpeechIPC thread is created lazily and kept as a static member on
the class. It is a LazyIdleThread: its serial event target is stable for the
process lifetime, so an actor is never handed a thread instance that later goes
away underneath it, but its backing OS thread is released once idle. The OS
thread is only ever idle while no connection is open (an active session streams
audio continuously), so it is never reclaimed mid-session.
- IPC only ever runs on this dedicated thread
- Static method calls are dispatched to that thread, and the result is
dispatched back to the main thread.
- In the next patch, processed and packetized audio is dispatched to that
thread as well
- Whenever a Speech Recognition session is active, or while an available or
install call is in flight, the HWInference connection is held open (refcounted)
- Otherwise the connection is closed as soon as we can (session stop, result of
`available` or `install`), so the utility process isn't kept alive by an idle
connection; the now-idle OS thread is then reclaimed by the LazyIdleThread

The logic to manage those transient sessions (vs. long running ones) is
encapsulated in a dedicated class.

Differential Revision: https://phabricator.services.mozilla.com/D275653

Author: padenot

build/clang-plugin/ThreadAllows.txt

@@ -25,6 +25,10 @@ UpdateProcessor
 BkgrndClipboard
 # Blocking thread calling into the parakeet.cpp library.
 Parakeet
+# Used by `SpeechRecognitionBackend` to send audio chunks and other commands to
+# the HWInference process. Needs to be a dedicated thread because some
+# scheduling guarantees are necessary.
+SpeechIPC
 
 ######
 # Thunderbird-only thread names

dom/media/webspeech/recognition/SpeechRecognitionBackend.cpp

@@ -6,24 +6,82 @@
 
 #include "SpeechRecognitionBackend.h"
 
+#include <utility>
+
 #include "SpeechRecognition.h"
 #include "SpeechTrackListener.h"
 #include "mozilla/AbstractThread.h"
 #include "mozilla/Assertions.h"
+#include "mozilla/ClearOnShutdown.h"
 #include "mozilla/dom/AudioStreamTrack.h"
 #include "mozilla/dom/ContentChild.h"
 #include "mozilla/dom/Promise.h"
+#include "mozilla/dom/SpeechRecognitionBinding.h"
+#include "mozilla/hwinference/PSpeechRecognition.h"
+#include "mozilla/hwinference/SpeechRecognitionChild.h"
+#include "mozilla/ipc/MessageChannel.h"
+#include "mozilla/ipc/ProtocolUtils.h"
 
 namespace mozilla::dom {
 
+using namespace mozilla::ipc;
+
+StaticAutoPtr<mozilla::EventTargetCapability<nsISerialEventTarget>>
+    SpeechRecognitionBackend::sIPCCapability;
+int32_t SpeechRecognitionBackend::sIPCActorUsers = 0;
+
 static LazyLogModule gSpeechRecognitionBackendLog("SpeechRecognitionBackend");
 
-#define LOG(...) \
-  MOZ_LOG_FMT(gSpeechRecognitionBackendLog, LogLevel::Debug, __VA_ARGS__)
-#define LOGV(...) \
-  MOZ_LOG_FMT(gSpeechRecognitionBackendLog, LogLevel::Verbose, __VA_ARGS__)
-#define LOGE(...) \
-  MOZ_LOG_FMT(gSpeechRecognitionBackendLog, LogLevel::Error, __VA_ARGS__)
+#define LOG(fmt, ...)                                                      \
+  MOZ_LOG_FMT(gSpeechRecognitionBackendLog, mozilla::LogLevel::Debug, fmt, \
+              ##__VA_ARGS__)
+#define LOGV(fmt, ...)                                                       \
+  MOZ_LOG_FMT(gSpeechRecognitionBackendLog, mozilla::LogLevel::Verbose, fmt, \
+              ##__VA_ARGS__)
+#define LOGE(fmt, ...)                                                     \
+  MOZ_LOG_FMT(gSpeechRecognitionBackendLog, mozilla::LogLevel::Error, fmt, \
+              ##__VA_ARGS__)
+
+// How long the shared IPC thread stays idle before its backing OS thread is
+// released. The serial event target itself lives for the process lifetime.
+static constexpr uint32_t IPC_THREAD_IDLE_TIMEOUT_MS = 5000;
+
+SpeechRecognitionBackend::SpeechRecognitionIPCActorUserGuard::
+    ~SpeechRecognitionIPCActorUserGuard() {
+  if (NS_IsMainThread()) {
+    SpeechRecognitionBackend::ReleaseIPCActorUser();
+  } else {
+    NS_DispatchToMainThread(NS_NewRunnableFunction(
+        "SpeechRecognitionIPCActorUserGuard::Release", [] {
+          AssertIsOnMainThread();
+          SpeechRecognitionBackend::ReleaseIPCActorUser();
+        }));
+  }
+}
+
+/* static */
+void SpeechRecognitionBackend::AcquireIPCActorUser() {
+  AssertIsOnMainThread();
+  ++sIPCActorUsers;
+}
+
+/* static */
+void SpeechRecognitionBackend::ReleaseIPCActorUser() {
+  AssertIsOnMainThread();
+  MOZ_ASSERT(sIPCActorUsers > 0);
+  if (--sIPCActorUsers || !sIPCCapability) {
+    return;
+  }
+
+  // Close the HWInference connection once no session needs it, so the utility
+  // process isn't kept alive by an idle connection. The serial event target is
+  // kept (its backing OS thread is released on idle by the LazyIdleThread);
+  // the next EnsureIPC() reopens the connection on that same target.
+  nsCOMPtr<nsIRunnable> close = NS_NewRunnableFunction(
+      "SpeechRecognitionBackend::CloseHWInferenceChildIfAny",
+      [] { CloseHWInferenceChildIfAny(); });
+  sIPCCapability->Dispatch(close.forget());
+}
 
 SpeechRecognitionBackend::SpeechRecognitionBackend(
     SpeechRecognition* aParent, uint32_t aGraphRate, const nsString& aLanguage,
@@ -56,14 +114,221 @@ void SpeechRecognitionBackend::DataCallback(TrackTime aTime,
 
 void SpeechRecognitionBackend::NotifyTrackEnded() {}
 
+/* static */
+void SpeechRecognitionBackend::EnsureIPCThread() {
+  AssertIsOnMainThread();
+
+  if (!sIPCCapability) {
+    // IPC actors are bound to the event target they were opened on, so the
+    // target has to outlive them: a LazyIdleThread keeps a single one for the
+    // process lifetime, and only releases its backing OS thread when idle.
+    RefPtr<LazyIdleThread> thread =
+        new LazyIdleThread(IPC_THREAD_IDLE_TIMEOUT_MS, "SpeechIPC");
+    sIPCCapability = new EventTargetCapability<nsISerialEventTarget>(thread);
+    LOG("Created shared IPC thread for speech recognition");
+    ClearOnShutdown(&sIPCCapability);
+  }
+}
+
+/* static */
+void SpeechRecognitionBackend::AssertOnIPCThread() {
+  sIPCCapability->AssertOnCurrentThread();
+}
+
+/* static */
+template <typename SendFunc>
+auto SpeechRecognitionBackend::RunWithTransientSession(SendFunc&& aSendFunc) {
+  AssertIsOnMainThread();
+
+  using SendPromise = typename decltype(aSendFunc(
+      std::declval<hwinference::SpeechRecognitionChild*>()))::element_type;
+  using ResolveValueType = typename SendPromise::ResolveValueType;
+  using OperationPromise = MozPromise<ResolveValueType, nsresult, true>;
+
+  MozPromiseHolder<OperationPromise> holder;
+  RefPtr<OperationPromise> operation = holder.Ensure(__func__);
+  CreateSession([holder = std::move(holder),
+                 aSendFunc = std::forward<SendFunc>(aSendFunc)](
+                    hwinference::SpeechRecognitionChild* aChild) mutable {
+    AssertOnIPCThread();
+    if (!aChild) {
+      holder.Reject(NS_ERROR_FAILURE, __func__);
+      return;
+    }
+    RefPtr child = aChild;
+    aSendFunc(aChild)->Then(
+        GetCurrentSerialEventTarget(), __func__,
+        [holder = std::move(holder),
+         child](typename SendPromise::ResolveOrRejectValue&& aValue) mutable {
+          AssertOnIPCThread();
+          child->Close();
+          if (aValue.IsReject()) {
+            holder.Reject(NS_ERROR_FAILURE, __func__);
+            return;
+          }
+          holder.Resolve(std::move(aValue.ResolveValue()), __func__);
+        });
+  });
+  return operation;
+}
+
+/* static */
 already_AddRefed<Promise> SpeechRecognitionBackend::Available(
     nsIGlobalObject* aGlobal, const nsTArray<nsCString>& aLanguages) {
-  return nullptr;
+  AssertIsOnMainThread();
+
+  if (!aGlobal) {
+    return nullptr;
+  }
+
+  ErrorResult rv;
+  RefPtr<Promise> promise = Promise::Create(aGlobal, rv);
+  if (rv.Failed()) {
+    return nullptr;
+  }
+
+  nsTArray<nsCString> languages = aLanguages.Clone();
+  if (languages.IsEmpty()) {
+    languages.AppendElement("en-US"_ns);
+  }
+
+  LOG("SpeechRecognitionBackend::Available - Starting availability check for "
+      "{} languages",
+      languages.Length());
+
+  if (MOZ_LOG_TEST(gSpeechRecognitionBackendLog, LogLevel::Debug)) {
+    for (const auto& lang : languages) {
+      LOG("SpeechRecognitionBackend::Available - Language requested: {}",
+          lang.get());
+    }
+  }
+
+  using IsModelInstalledPromise =
+      hwinference::PSpeechRecognitionChild::IsModelInstalledPromise;
+  using AvailabilityPromise = MozPromise<AvailabilityStatus, nsresult, true>;
+  // RunWithTransientSession turns any rejection into NS_ERROR_FAILURE, so the
+  // IPC reject reason is forwarded verbatim rather than converted at each step.
+  using SendAvailabilityPromise =
+      MozPromise<AvailabilityStatus, ResponseRejectReason, true>;
+  RunWithTransientSession(
+      [languages = std::move(languages)](
+          hwinference::SpeechRecognitionChild* aChild) mutable
+          -> RefPtr<SendAvailabilityPromise> {
+        RefPtr<hwinference::SpeechRecognitionChild> child = aChild;
+        return IsModelInstalledNative(child, languages)
+            ->Then(GetCurrentSerialEventTarget(), __func__,
+                   [child, languages = std::move(languages)](
+                       IsModelInstalledPromise::ResolveOrRejectValue&& aValue)
+                       -> RefPtr<SendAvailabilityPromise> {
+                     if (aValue.IsReject()) {
+                       return SendAvailabilityPromise::CreateAndReject(
+                           aValue.RejectValue(), __func__);
+                     }
+                     if (aValue.ResolveValue()) {
+                       return SendAvailabilityPromise::CreateAndResolve(
+                           AvailabilityStatus::Available, __func__);
+                     }
+                     return child->SendIsModelAvailable(languages)
+                         ->Map(GetCurrentSerialEventTarget(), __func__,
+                               [](bool aAvailable) {
+                                 return aAvailable
+                                            ? AvailabilityStatus::Downloadable
+                                            : AvailabilityStatus::Unavailable;
+                               });
+                   });
+      })
+      ->Then(GetMainThreadSerialEventTarget(), __func__,
+             [promise](AvailabilityPromise::ResolveOrRejectValue&& aValue) {
+               promise->MaybeResolve(aValue.IsResolve()
+                                         ? aValue.ResolveValue()
+                                         : AvailabilityStatus::Unavailable);
+             });
+
+  return promise.forget();
+}
+
+/* static */
+RefPtr<hwinference::PSpeechRecognitionChild::IsModelInstalledPromise>
+SpeechRecognitionBackend::IsModelInstalledNative(
+    hwinference::SpeechRecognitionChild* aChild,
+    const nsTArray<nsCString>& aLanguages) {
+  AssertOnIPCThread();
+
+  LOG("SpeechRecognitionBackend::IsModelInstalledNative - Starting installed "
+      "check for {} languages",
+      aLanguages.Length());
+
+  return aChild->SendIsModelInstalled(aLanguages);
+}
+
+/* static */
+void SpeechRecognitionBackend::CreateSession(
+    MoveOnlyFunction<void(hwinference::SpeechRecognitionChild*)> aCallback) {
+  AssertIsOnMainThread();
+
+  RefPtr<SpeechRecognitionIPCActorUserGuard> guard =
+      MakeRefPtr<SpeechRecognitionIPCActorUserGuard>();
+  EnsureIPCThread();
+
+  Endpoint<hwinference::PSpeechRecognitionParent> parentEndpoint;
+  Endpoint<hwinference::PSpeechRecognitionChild> childEndpoint;
+  MOZ_ALWAYS_SUCCEEDS(hwinference::PSpeechRecognition::CreateEndpoints(
+      &parentEndpoint, &childEndpoint));
+  ContentChild::GetSingleton()->SendCreateSpeechRecognition(
+      std::move(parentEndpoint));
+
+  sIPCCapability->Dispatch(NS_NewRunnableFunction(
+      "SpeechRecognitionBackend::CreateSession",
+      [guard = std::move(guard), endpoint = std::move(childEndpoint),
+       callback = std::move(aCallback)]() mutable {
+        AssertOnIPCThread();
+        RefPtr child = new hwinference::SpeechRecognitionChild(guard.forget());
+        if (!endpoint.Bind(child)) {
+          callback(nullptr);
+          return;
+        }
+        callback(child);
+      }));
 }
 
+/* static */
 already_AddRefed<Promise> SpeechRecognitionBackend::Install(
     nsIGlobalObject* aGlobal, const nsTArray<nsCString>& aLanguages) {
-  return nullptr;
+  AssertIsOnMainThread();
+
+  if (!aGlobal) {
+    return nullptr;
+  }
+
+  ErrorResult rv;
+  RefPtr<Promise> promise = Promise::Create(aGlobal, rv);
+  if (rv.Failed()) {
+    return nullptr;
+  }
+
+  if (aLanguages.IsEmpty()) {
+    promise->MaybeResolve(false);
+    return promise.forget();
+  }
+
+  LOG("SpeechRecognitionBackend::Install - Starting install for {} languages",
+      aLanguages.Length());
+
+  using InstallPromise = MozPromise<bool, nsresult, true>;
+  RunWithTransientSession(
+      [languages = aLanguages.Clone()](
+          hwinference::SpeechRecognitionChild* aChild) mutable {
+        return aChild->SendInstallModels(std::move(languages));
+      })
+      ->Then(GetMainThreadSerialEventTarget(), __func__,
+             [promise](InstallPromise::ResolveOrRejectValue&& aValue) {
+               bool success = aValue.IsResolve() && aValue.ResolveValue();
+               LOG("SpeechRecognitionBackend::Install - Install completed: {}",
+                   success ? "success" : "failed");
+               promise->MaybeResolve(success);
+             });
+
+  return promise.forget();
 }
 
 }  // namespace mozilla::dom