broadcastradio: Add Audio Generator and PCM Stream Server for Virtual Tuner

base/cvd/cuttlefish/common/libs/fs/shared_buf.cc

@@ -74,14 +74,15 @@ ssize_t WriteAll(SharedFD fd, const std::vector<char>& buf) {
   return WriteAll(fd, buf.data(), buf.size());
 }
 
-bool SendAll(SharedFD sock, std::string_view msg) {
+bool SendAll(SharedFD sock, const void* buf, size_t size, int flags) {
   ssize_t total_written{};
   if (!sock->IsOpen()) {
     return false;
   }
-  while (total_written < static_cast<ssize_t>(msg.size())) {
-    auto just_written = sock->Send(msg.data() + total_written,
-                                   msg.size() - total_written, MSG_NOSIGNAL);
+  const char* cursor = static_cast<const char*>(buf);
+  while (total_written < static_cast<ssize_t>(size)) {
+    ssize_t just_written =
+        sock->Send(cursor + total_written, size - total_written, flags);
     if (just_written <= 0) {
       return false;
     }
@@ -90,4 +91,8 @@ bool SendAll(SharedFD sock, std::string_view msg) {
   return true;
 }
 
+bool SendAll(SharedFD sock, std::string_view msg, int flags) {
+  return SendAll(sock, msg.data(), msg.size(), flags);
+}
+
 }  // namespace cuttlefish

base/cvd/cuttlefish/common/libs/fs/shared_buf.h

@@ -148,6 +148,17 @@ ssize_t WriteAllBinary(SharedFD fd, const T* binary_data) {
   return WriteAll(fd, (const char*)binary_data, sizeof(*binary_data));
 }
 
+/**
+ * Sends contents of buf through sock, checking for socket error conditions
+ *
+ * On successful Send, returns true
+ *
+ * If a Send error is encountered, returns false. Some data may have already
+ * been written to 'sock' at that point.
+ */
+bool SendAll(SharedFD sock, const void* buf, size_t size,
+             int flags = MSG_NOSIGNAL);
+
 /**
  * Sends contents of msg through sock, checking for socket error conditions
  *
@@ -156,6 +167,6 @@ ssize_t WriteAllBinary(SharedFD fd, const T* binary_data) {
  * If a Send error is encountered, returns false. Some data may have already
  * been written to 'sock' at that point.
  */
-bool SendAll(SharedFD sock, std::string_view msg);
+bool SendAll(SharedFD sock, std::string_view msg, int flags = MSG_NOSIGNAL);
 
 }  // namespace cuttlefish

base/cvd/cuttlefish/host/commands/virtual_tuner_daemon/BUILD.bazel

@@ -50,3 +50,36 @@ cf_cc_test(
         ":virtual_tuner_cc_proto",
     ],
 )
+
+cf_cc_library(
+    name = "virtual_tuner_audio",
+    srcs = [
+        "audio_generator.cpp",
+        "pcm_stream_server.cpp",
+    ],
+    hdrs = [
+        "audio_generator.h",
+        "pcm_stream_server.h",
+    ],
+    deps = [
+        ":tuner_state",
+        "//cuttlefish/common/libs/fs",
+        "//cuttlefish/common/libs/fs:fd",
+        "//cuttlefish/result",
+        "@abseil-cpp//absl/log",
+    ],
+)
+
+cf_cc_test(
+    name = "virtual_tuner_audio_test",
+    srcs = [
+        "virtual_tuner_audio_test.cpp",
+    ],
+    deps = [
+        ":tuner_state",
+        ":virtual_tuner_audio",
+        ":virtual_tuner_cc_proto",
+        "//cuttlefish/common/libs/fs",
+        "//cuttlefish/result",
+    ],
+)

base/cvd/cuttlefish/host/commands/virtual_tuner_daemon/audio_generator.cpp

@@ -0,0 +1,55 @@
+/*
+ * Copyright (C) 2026 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "cuttlefish/host/commands/virtual_tuner_daemon/audio_generator.h"
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <random>
+#include <span>
+
+#include "cuttlefish/host/commands/virtual_tuner_daemon/tuner_state.h"
+
+namespace cuttlefish {
+namespace virtualtuner {
+namespace {
+
+constexpr int16_t kNoiseAmplitude = 4000;
+
+}  // namespace
+
+AudioGenerator::AudioGenerator() : rng_(std::random_device()()) {}
+
+void AudioGenerator::GenerateChunk(std::span<int16_t> samples,
+                                   const TunerStateSnapshot& snapshot) {
+  if (!snapshot.is_playing) {
+    std::fill(samples.begin(), samples.end(), 0);
+    return;
+  }
+
+  std::uniform_int_distribution<int> noise(-kNoiseAmplitude, kNoiseAmplitude);
+  for (size_t frame = 0; frame + kChannels <= samples.size();
+       frame += kChannels) {
+    const auto sample = static_cast<int16_t>(noise(rng_));
+    for (size_t channel = 0; channel < kChannels; ++channel) {
+      samples[frame + channel] = sample;
+    }
+  }
+}
+
+}  // namespace virtualtuner
+}  // namespace cuttlefish

base/cvd/cuttlefish/host/commands/virtual_tuner_daemon/audio_generator.h

@@ -0,0 +1,49 @@
+/*
+ * Copyright (C) 2026 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <random>
+#include <span>
+
+#include "cuttlefish/host/commands/virtual_tuner_daemon/tuner_state.h"
+
+namespace cuttlefish {
+namespace virtualtuner {
+
+inline constexpr size_t kSampleRate = 48000;
+inline constexpr size_t kChannels = 2;
+inline constexpr size_t kBytesPerSample = sizeof(int16_t);
+inline constexpr size_t kFrameSizeBytes = kChannels * kBytesPerSample;
+
+inline constexpr size_t kChunkFrames = 4096;
+inline constexpr size_t kChunkSizeBytes = kChunkFrames * kFrameSizeBytes;
+
+class AudioGenerator {
+ public:
+  AudioGenerator();
+
+  void GenerateChunk(std::span<int16_t> samples,
+                     const TunerStateSnapshot& snapshot);
+
+ private:
+  std::minstd_rand rng_;
+};
+
+}  // namespace virtualtuner
+}  // namespace cuttlefish

base/cvd/cuttlefish/host/commands/virtual_tuner_daemon/pcm_stream_server.cpp

@@ -0,0 +1,196 @@
+/*
+ * Copyright (C) 2026 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "cuttlefish/host/commands/virtual_tuner_daemon/pcm_stream_server.h"
+
+#include <atomic>
+#include <cerrno>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <ratio>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include "absl/log/log.h"
+
+#include "cuttlefish/common/libs/fs/fd.h"
+#include "cuttlefish/common/libs/fs/shared_buf.h"
+#include "cuttlefish/common/libs/fs/shared_fd.h"
+#include "cuttlefish/host/commands/virtual_tuner_daemon/audio_generator.h"
+#include "cuttlefish/host/commands/virtual_tuner_daemon/tuner_state.h"
+#include "cuttlefish/result/result.h"
+
+namespace cuttlefish {
+namespace virtualtuner {
+namespace {
+
+using ChunkDuration =
+    std::chrono::duration<int64_t, std::ratio<kChunkFrames, kSampleRate>>;
+
+constexpr auto kMaxSchedulingLag = std::chrono::milliseconds(500);
+constexpr int kPrebufferChunks = 2;
+
+}  // namespace
+
+PcmStreamServer::PcmStreamServer(TunerState& tuner_state, SharedFD server_fd)
+    : tuner_state_(tuner_state), server_fd_(std::move(server_fd)) {}
+
+PcmStreamServer::~PcmStreamServer() { Stop(); }
+
+Result<void> PcmStreamServer::Start() {
+  if (is_running_) {
+    return {};
+  }
+
+  CF_EXPECT(server_fd_->IsOpen(),
+            "Failed to start PCM streaming server: server_fd is not open");
+
+  LOG(INFO) << "PCM streaming server running with inherited server fd.";
+  is_running_ = true;
+  accept_thread_ = std::thread(&PcmStreamServer::AcceptLoop, this);
+  return {};
+}
+
+void PcmStreamServer::Stop() {
+  if (!is_running_.exchange(false)) {
+    return;
+  }
+
+  if (server_fd_->IsOpen()) {
+    server_fd_->Shutdown(SHUT_RDWR);
+    server_fd_->Close();
+  }
+  stop_cv_.notify_all();
+  if (accept_thread_.joinable()) {
+    accept_thread_.join();
+  }
+
+  std::vector<std::unique_ptr<ClientSession>> sessions;
+  {
+    std::lock_guard<std::mutex> lock(clients_mutex_);
+    sessions.swap(clients_);
+  }
+  for (const auto& session : sessions) {
+    if (session->fd->IsOpen()) {
+      session->fd->Shutdown(SHUT_RDWR);
+      session->fd->Close();
+    }
+  }
+  for (auto& session : sessions) {
+    if (session->thread.joinable()) {
+      session->thread.join();
+    }
+  }
+
+  LOG(INFO) << "PCM streaming server stopped.";
+}
+
+void PcmStreamServer::AcceptLoop() {
+  while (is_running_) {
+    SharedFD client_fd = Fd::Accept(*server_fd_).value_or(Fd());
+    if (!is_running_) {
+      break;
+    }
+    if (!client_fd->IsOpen()) {
+      LOG(ERROR) << "Accept failed on PCM server socket: "
+                 << client_fd->StrError();
+      continue;
+    }
+
+    std::lock_guard<std::mutex> lock(clients_mutex_);
+    ReapFinishedClients();
+    if (clients_.size() >= kMaxConcurrentClients) {
+      LOG(WARNING) << "Refusing PCM client: already serving " << clients_.size()
+                   << " clients.";
+      continue;
+    }
+
+    LOG(INFO) << "PCM streaming server accepted a client.";
+    auto session = std::make_unique<ClientSession>();
+    session->fd = client_fd;
+    ClientSession* session_ptr = session.get();
+    clients_.push_back(std::move(session));
+    session_ptr->thread = std::thread(&PcmStreamServer::StreamClient, this,
+                                      std::ref(*session_ptr));
+  }
+}
+
+void PcmStreamServer::ReapFinishedClients() {
+  std::erase_if(clients_, [](const std::unique_ptr<ClientSession>& session) {
+    if (!session->finished.load(std::memory_order_acquire)) {
+      return false;
+    }
+    if (session->thread.joinable()) {
+      session->thread.join();
+    }
+    return true;
+  });
+}
+
+void PcmStreamServer::StreamClient(ClientSession& session) {
+  const SharedFD& client_fd = session.fd;
+  AudioGenerator generator;
+  std::vector<int16_t> buffer(kChunkFrames * kChannels);
+
+  const auto write_next_chunk = [&]() -> bool {
+    generator.GenerateChunk(buffer, tuner_state_.GetSnapshot());
+    return SendAll(client_fd, buffer.data(), buffer.size() * sizeof(int16_t));
+  };
+
+  bool connected = true;
+  for (int i = 0; i < kPrebufferChunks && connected && is_running_; ++i) {
+    connected = write_next_chunk();
+  }
+
+  auto anchor = std::chrono::steady_clock::now();
+  int64_t chunks_since_anchor = 0;
+
+  while (connected && is_running_) {
+    ++chunks_since_anchor;
+    const auto deadline = anchor + ChunkDuration(chunks_since_anchor);
+    {
+      std::unique_lock<std::mutex> lock(stop_mutex_);
+      if (stop_cv_.wait_until(lock, deadline,
+                              [&]() -> bool { return !is_running_; })) {
+        break;
+      }
+    }
+
+    const auto now = std::chrono::steady_clock::now();
+    if (now > deadline + kMaxSchedulingLag) {
+      LOG(WARNING) << "PCM stream fell behind schedule by "
+                   << std::chrono::duration_cast<std::chrono::milliseconds>(
+                          now - deadline)
+                          .count()
+                   << " ms; re-anchoring.";
+      anchor = now;
+      chunks_since_anchor = 0;
+    }
+
+    connected = write_next_chunk();
+  }
+
+  LOG(INFO) << "PCM client disconnected.";
+  session.finished.store(true, std::memory_order_release);
+}
+
+}  // namespace virtualtuner
+}  // namespace cuttlefish