server/render.c

Go to the documentation of this file.

 
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <errno.h>
#include <math.h>
 
#include "render.h"
#include "client.h"
#include "main.h"
#include "stream.h"
#include "protocol.h"
#include <ascii-chat/common.h>
#include <ascii-chat/app_callbacks.h>
#include <ascii-chat/options/options.h>
#include <ascii-chat/options/rcu.h> // For RCU-based options access
#include <ascii-chat/platform/abstraction.h>
#include <ascii-chat/platform/init.h>
#include <ascii-chat/network/packet/queue.h>
#include <ascii-chat/network/acip/server.h>
#include <ascii-chat/util/time.h>
#include <ascii-chat/audio/mixer.h>
#include <ascii-chat/audio/audio.h>
#include <ascii-chat/audio/opus.h>
#include <ascii-chat/util/format.h>
#include <ascii-chat/util/fps.h>
 
/* ============================================================================
 * Cross-Platform Utility Functions
 * ============================================================================
 */
 
// Removed interruptible_usleep - using regular platform_sleep_us instead
// Sleep interruption isn't needed for small delays and isn't truly possible anyway
 
/* ============================================================================
 * Per-Client Video Rendering Implementation
 * ============================================================================
 */
 
void *client_video_render_thread(void *arg) {
  client_info_t *client = (client_info_t *)arg;
  if (!client) {
    log_error("NULL client pointer in video render thread");
    return NULL;
  }
 
  // Take snapshot of client ID and socket at start to avoid race conditions
  char thread_client_id[MAX_CLIENT_ID_LEN];
  SAFE_STRNCPY(thread_client_id, client->client_id, sizeof(thread_client_id) - 1);
  socket_t thread_socket = client->socket;
  bool is_webrtc = (thread_socket == INVALID_SOCKET_VALUE);
  (void)is_webrtc; // May be unused in release builds
 
  log_debug("Video render thread: client_id=%s, webrtc=%d", thread_client_id, is_webrtc);
 
  log_info("[VIDEO_RENDER_THREAD_START] ★★★ LOCK STATE at thread entry for client %s", thread_client_id);
  // debug_sync_print_state();  // Disabled: causes AddressSanitizer stack-use-after-return crash
 
  // Wait for client to send terminal capabilities before rendering
  // Without capabilities, convert_composite_to_ascii() will fail
  int timeout_ms = 5000;
  int waited_ms = 0;
  while (!client->has_terminal_caps && !atomic_load_bool(&g_should_exit) && !atomic_load_bool(&client->shutting_down)) {
    if (waited_ms == 0) {
      log_debug("Waiting for terminal capabilities from client %s...", thread_client_id);
    }
    APP_CALLBACK_VOID(platform_pump_events);
    platform_sleep_ms(10);
    waited_ms += 10;
    if (waited_ms >= timeout_ms) {
      log_debug("Still waiting for terminal capabilities from client %s", thread_client_id);
      waited_ms = 0;
    }
  }
  if (client->has_terminal_caps) {
    log_debug("Received terminal capabilities for client %s after %dms", thread_client_id, waited_ms);
  }
 
  // Get client's desired FPS from capabilities or use default
  int client_fps = VIDEO_RENDER_FPS; // Default to 60 FPS
  // Use snapshot pattern to avoid mutex in render thread
  bool has_caps = client->has_terminal_caps;
  int desired_fps = has_caps ? client->terminal_caps.desired_fps : 0;
  if (has_caps && desired_fps > 0) {
    client_fps = desired_fps;
    log_debug("Client %s requested FPS: %d (has_caps=%d, desired_fps=%d)", thread_client_id, client_fps, has_caps,
              desired_fps);
  } else {
    log_debug("Client %s using default FPS: %d (has_caps=%d, desired_fps=%d)", thread_client_id, client_fps, has_caps,
              desired_fps);
  }
 
  int base_frame_interval_ms = 1000 / client_fps;
  log_debug("Client %s render interval: %dms (%d FPS)", thread_client_id, base_frame_interval_ms, client_fps);
 
  // FPS tracking for video render thread
  fps_t video_fps_tracker = {0};
  fps_init(&video_fps_tracker, client_fps, "SERVER VIDEO");
 
  const uint64_t frame_interval_ns = NS_PER_SEC_INT / client_fps;
  uint64_t next_frame_ns = time_get_ns() + frame_interval_ns;
  bool last_has_sources = false;
  uint32_t frame_gen_count = 0;
  uint64_t frame_gen_start_time = 0;
  uint32_t last_frame_hash = UINT32_MAX;
  uint32_t commits_count = 0;
  uint64_t commits_start_time = 0;
  uint64_t render_rate_window_start_ns = time_get_ns();
  uint64_t rendered_frames_in_window = 0;
  uint64_t missed_deadlines_in_window = 0;
 
  log_info("Video render loop STARTING for client %s", thread_client_id);
 
  bool should_continue = true;
  while (should_continue && !atomic_load_bool(&g_should_exit) && !atomic_load_bool(&client->shutting_down)) {
    START_TIMER("render_iteration");
    log_dev_every(10 * NS_PER_MS_INT, "Video render loop iteration for client %s", thread_client_id);
 
    // Check for immediate shutdown
    if (atomic_load_bool(&g_should_exit)) {
      log_debug("Video render thread stopping for client %s (g_should_exit)", thread_client_id);
      break;
    }
 
    bool video_running = atomic_load_bool(&client->video_render_thread_running);
    bool active = atomic_load_bool(&client->active);
    bool shutting_down = atomic_load_bool(&client->shutting_down);
 
    should_continue = video_running && active && !shutting_down;
 
    if (!should_continue) {
      log_debug("Video render thread stopping for client %s (should_continue=false: video_running=%d, active=%d, "
                "shutting_down=%d)",
                thread_client_id, video_running, active, shutting_down);
      break;
    }
 
    // Processing consumes part of the frame interval. Sleep only until the
    // scheduled deadline, and skip missed deadlines without a catch-up burst.
    uint64_t iter_start_ns = time_get_ns();
    START_TIMER("render_frame_sleep");
    if (iter_start_ns < next_frame_ns) {
      platform_sleep_ns(next_frame_ns - iter_start_ns);
    }
    STOP_TIMER_AND_LOG(dev, 0, "render_frame_sleep", "frame deadline sleep completed");
    uint64_t after_sleep_ns = time_get_ns();
    uint64_t elapsed_intervals =
        after_sleep_ns >= next_frame_ns ? (after_sleep_ns - next_frame_ns) / frame_interval_ns + 1 : 1;
    if (elapsed_intervals > 1) {
      missed_deadlines_in_window += elapsed_intervals - 1;
    }
    next_frame_ns += elapsed_intervals * frame_interval_ns;
 
    // Capture timestamp for FPS tracking and frame timestamps
    uint64_t current_time_ns = time_get_ns();
 
    // Check thread state again before acquiring locks (client might have been destroyed during sleep).
    should_continue = atomic_load_bool(&client->video_render_thread_running) && atomic_load_bool(&client->active) &&
                      !atomic_load_bool(&client->shutting_down);
    if (!should_continue) {
      break;
    }
 
    // Optimization: No mutex needed - all fields are atomic or stable.
    // client_id: stable string pointer
    // width/height: stable uint16_t values (set once, never modified)
    // active: atomic_t - use atomic_load for thread safety
    const char *client_id_snapshot = client->client_id;       // Stable read
    unsigned short width_snapshot = client->width;            // Stable uint16_t
    unsigned short height_snapshot = client->height;          // Stable uint16_t
    bool active_snapshot = atomic_load_bool(&client->active); // Atomic read
 
    // Check if client is still active after getting snapshot
    if (!active_snapshot) {
      break;
    }
 
    // Phase 2 IMPLEMENTED: Generate frame specifically for THIS client using snapshot data
    size_t frame_size = 0;
 
    // Check if any clients are sending video
    bool has_video_sources = any_clients_sending_video();
 
    // DIAGNOSTIC: Track when video sources become available
    if (has_video_sources != last_has_sources) {
      log_warn("DIAGNOSTIC: Client %s video sources: %s", thread_client_id,
               has_video_sources ? "AVAILABLE" : "UNAVAILABLE");
      last_has_sources = has_video_sources;
    }
 
    log_debug_every(5 * NS_PER_MS_INT,
                    "Video render iteration for client %s: has_video_sources=%d, width=%u, height=%u", thread_client_id,
                    has_video_sources, width_snapshot, height_snapshot);
 
    // Use default dimensions if client dimensions not received
    if (width_snapshot == 0 || height_snapshot == 0) {
      log_dev_every(5 * NS_PER_MS_INT, "Using default dimensions for client %s (client reported: width=%u, height=%u)",
                    thread_client_id, width_snapshot, height_snapshot);
      // Use 80x25 as standard terminal size fallback for snapshot/test clients
      width_snapshot = 80;
      height_snapshot = 25;
    }
 
    // Generate frames for all clients with valid dimensions, regardless of whether other clients are sending video.
    // This ensures WebSocket snapshot clients (receive-only mode) receive frames even when no other clients are
    // sending video.
    int sources_count = 0; // Track number of video sources in this frame
 
    // DIAGNOSTIC: Track every frame generation attempt
    frame_gen_count++;
    if (frame_gen_count == 1) {
      frame_gen_start_time = current_time_ns;
    }
 
    // Log every 120 attempts (should be ~2 seconds at 60 Hz)
    if (frame_gen_count % 120 == 0) {
      uint64_t elapsed_ns = current_time_ns - frame_gen_start_time;
      double gen_fps = (120.0 / (elapsed_ns / (double)NS_PER_SEC_INT));
      log_dev("render loop: client=%s fps=%.1f", thread_client_id, gen_fps);
    }
 
    log_dev_every(5 * NS_PER_MS_INT, "About to call create_mixed_ascii_frame_for_client for client %s with dims %ux%u",
                  thread_client_id, width_snapshot, height_snapshot);
 
    // debug_sync_print_state() is too expensive to call every frame (kills FPS)
 
    uint64_t frame_create_start_ns = time_get_ns();
    START_TIMER("render_create_frame");
    char *ascii_frame = create_mixed_ascii_frame_for_client(client_id_snapshot, width_snapshot, height_snapshot, false,
                                                            &frame_size, NULL, &sources_count);
    STOP_TIMER_AND_LOG(dev, 0, "render_create_frame", "create_mixed_ascii_frame completed");
    uint64_t frame_create_end_ns = time_get_ns();
 
    if (frame_gen_count % 120 == 0) {
      char sleep_str[32], create_str[32];
      time_pretty((uint64_t)(after_sleep_ns - iter_start_ns), -1, sleep_str, sizeof(sleep_str));
      time_pretty((uint64_t)(frame_create_end_ns - frame_create_start_ns), -1, create_str, sizeof(create_str));
      log_dev("render timing: adaptive_sleep=%s frame_create=%s", sleep_str, create_str);
    }
 
    // Always send frames at configured FPS, even if no external video sources yet
    // This ensures single-client mode (like testing) still gets continuous frames
    // The frame data (animated background or test pattern) is still valid content
 
    // DEBUG: Log frame generation details
    uint32_t current_frame_hash = 0;
    if (ascii_frame && frame_size > 0) {
      for (size_t i = 0; i < frame_size && i < 1000; i++) {
        current_frame_hash = (uint32_t)((uint64_t)current_frame_hash * 31 + ((unsigned char *)ascii_frame)[i]);
      }
      if (current_frame_hash != last_frame_hash) {
        log_dev("RENDER_FRAME CHANGE: Client %s frame #%zu sources=%d hash=0x%08x (prev=0x%08x)", thread_client_id,
                 frame_size, sources_count, current_frame_hash, last_frame_hash);
        last_frame_hash = current_frame_hash;
      } else {
        log_dev_every(25000, "RENDER_FRAME DUPLICATE: Client %s frame #%zu sources=%d hash=0x%08x (no change)",
                      thread_client_id, frame_size, sources_count, current_frame_hash);
      }
    }
 
    log_dev_every(5 * NS_PER_MS_INT,
                  "create_mixed_ascii_frame_for_client returned: ascii_frame=%p, frame_size=%zu, sources_count=%d",
                  (void *)ascii_frame, frame_size, sources_count);
 
    // Phase 2 IMPLEMENTED: Write frame to double buffer (never drops!)
    if (ascii_frame && frame_size > 0) {
      log_debug_every(5 * NS_PER_MS_INT, "Buffering frame for client %s (size=%zu)", thread_client_id, frame_size);
      // GRID LAYOUT CHANGE DETECTION: Store source count with frame
      // Send thread will compare this with last sent count to detect grid changes
      atomic_store_u64(&client->last_rendered_grid_sources, (uint64_t)sources_count);
 
      // Use double-buffer system which has its own internal swap_mutex
      // No external locking needed - the double-buffer is thread-safe by design
      video_frame_buffer_t *vfb_snapshot = client->outgoing_video_buffer;
 
      if (vfb_snapshot) {
        video_frame_t *write_frame = video_frame_begin_write(vfb_snapshot);
        if (write_frame) {
          // Copy ASCII frame data to the back buffer (NOT holding rwlock - just double-buffer's internal lock)
          if (write_frame->data && frame_size <= vfb_snapshot->allocated_buffer_size) {
            memcpy(write_frame->data, ascii_frame, frame_size);
            write_frame->size = frame_size;
            write_frame->width = width_snapshot;
            write_frame->height = height_snapshot;
            write_frame->capture_timestamp_ns = current_time_ns;
 
            // Commit at the configured cadence enforced by the frame deadline.
            // Never skip frames based on content changes.
            {
              uint64_t commit_start_ns = time_get_ns();
              // Commit the frame (swaps buffers atomically using vfb->swap_mutex, NOT rwlock)
              START_TIMER("render_video_frame_commit");
              video_frame_commit(vfb_snapshot);
              STOP_TIMER_AND_LOG(dev, 0, "render_video_frame_commit", "video_frame_commit completed");
              uint64_t commit_end_ns = time_get_ns();
              char commit_duration_str[32];
              time_pretty((uint64_t)(commit_end_ns - commit_start_ns), -1, commit_duration_str,
                          sizeof(commit_duration_str));
 
              commits_count++;
              if (commits_count == 1) {
                commits_start_time = commit_end_ns;
              }
              if (commits_count % 10 == 0) {
                uint64_t elapsed_ns = commit_end_ns - commits_start_time;
                double commit_fps = (10.0 / (elapsed_ns / (double)NS_PER_SEC_INT));
                log_dev("render commit rate: client=%s fps=%.1f", thread_client_id, commit_fps);
              }
 
              log_dev("FRAME_COMMIT: Client %s took %s (hash=0x%08x)", thread_client_id,
                       commit_duration_str, current_frame_hash);
 
              // ARCHITECTURE: Frame transmission is handled EXCLUSIVELY by the send thread
              // The send thread reads from this buffer and applies proper 60 FPS rate limiting
              // Do NOT send from render thread - it bypasses rate limiting and causes network congestion
              // Render thread responsibility: Generate frame → Write to buffer → Done
              // Send thread responsibility: Read from buffer → Apply 60 FPS pacing → Transmit
            }
 
          } else {
            log_warn("Frame too large for buffer: %zu > %zu", frame_size, vfb_snapshot->allocated_buffer_size);
          }
 
          // FPS tracking - frame successfully generated (handles lag detection and periodic reporting)
          fps_frame_ns(&video_fps_tracker, current_time_ns, "frame rendered");
          rendered_frames_in_window++;
          uint64_t render_rate_now_ns = time_get_ns();
          uint64_t render_rate_elapsed_ns = render_rate_now_ns - render_rate_window_start_ns;
          if (render_rate_elapsed_ns >= NS_PER_SEC_INT) {
            double render_fps = (double)rendered_frames_in_window * NS_PER_SEC_INT / (double)render_rate_elapsed_ns;
            if (render_fps < (double)client_fps * 0.9 || missed_deadlines_in_window > 0) {
              log_warn("Server video render lagging client=%s fps=%.1f target=%d missed_deadlines=%llu",
                       thread_client_id, render_fps, client_fps,
                       (unsigned long long)missed_deadlines_in_window);
            } else {
              log_info("Server video render client=%s fps=%.1f target=%d missed_deadlines=0", thread_client_id,
                       render_fps, client_fps);
            }
            render_rate_window_start_ns = render_rate_now_ns;
            rendered_frames_in_window = 0;
            missed_deadlines_in_window = 0;
          }
        }
      }
 
      SAFE_FREE(ascii_frame);
    } else {
      // No frame generated (probably no video sources) - this is normal, no error logging needed
      log_dev_every(10 * NS_PER_MS_INT, "Per-client render: No video sources available for client %s",
                    client_id_snapshot);
    }
 
    uint64_t iter_end_ns = time_get_ns();
    if (frame_gen_count % 120 == 0) {
      char total_str[32];
      time_pretty((uint64_t)(iter_end_ns - iter_start_ns), -1, total_str, sizeof(total_str));
      log_warn("  TIMING: TOTAL_ITERATION=%s", total_str);
    }
    STOP_TIMER_AND_LOG(dev, 0, "render_iteration", "render_iteration completed (total)");
  }
 
#ifdef DEBUG_THREADS
  log_debug("Video render thread stopped for client %s", thread_client_id);
#endif
 
  // Clean up thread-local error context before exit
  asciichat_errno_destroy();
 
  return NULL;
}
 
/* ============================================================================
 * Per-Client Audio Rendering Implementation
 * ============================================================================
 */
 
void *client_audio_render_thread(void *arg) {
  client_info_t *client = (client_info_t *)arg;
 
  if (!client) {
    log_error("Invalid client info in audio render thread");
    return NULL;
  }
 
  // Take snapshot of client ID and display name at start to avoid race conditions
  char thread_client_id[MAX_CLIENT_ID_LEN];
  SAFE_STRNCPY(thread_client_id, client->client_id, sizeof(thread_client_id) - 1);
  char thread_display_name[64];
  bool is_webrtc = (client->socket == INVALID_SOCKET_VALUE);
  (void)is_webrtc; // May be unused in release builds
 
  // LOCK OPTIMIZATION: Only need client_state_mutex, not global rwlock
  // We already have a stable client pointer
  mutex_lock(&client->client_state_mutex);
  SAFE_STRNCPY(thread_display_name, client->display_name, sizeof(thread_display_name));
  mutex_unlock(&client->client_state_mutex);
 
#ifdef DEBUG_THREADS
  log_debug("Audio render thread started for client %s (%s), webrtc=%d", thread_client_id, thread_display_name,
            is_webrtc);
#endif
 
  // Mix buffer: up to 960 samples for adaptive reading
  // Normal: 480 samples = 10ms @ 48kHz
  // Catchup: 960 samples = 20ms when buffers are filling up
  float mix_buffer[960];
 
// Opus frame accumulation buffer (960 samples = 20ms @ 48kHz)
// Opus requires minimum 480 samples, 960 is optimal for 20ms frames
#define OPUS_FRAME_SAMPLES 960
  float opus_frame_buffer[OPUS_FRAME_SAMPLES];
  int opus_frame_accumulated = 0;
 
  // Create Opus encoder for this client's audio stream (48kHz, mono, 128kbps, AUDIO mode for music quality)
  opus_codec_t *opus_encoder = opus_codec_create_encoder(OPUS_APPLICATION_AUDIO, 48000, 128000);
  if (!opus_encoder) {
    log_error("Failed to create Opus encoder for audio render thread (client %s)", thread_client_id);
    return NULL;
  }
 
  // FPS tracking for audio render thread
  fps_t audio_fps_tracker = {0};
  fps_init(&audio_fps_tracker, AUDIO_PACKET_FPS, "SERVER AUDIO");
 
  // Per-thread counters (NOT static - each thread instance gets its own)
  int server_audio_frame_count = 0;
 
  bool should_continue = true;
  uint64_t audio_deadline = time_get_ns();
  while (should_continue && !atomic_load_bool(&g_should_exit) && !atomic_load_bool(&client->shutting_down)) {
    log_debug_every(LOG_RATE_SLOW, "Audio render loop iteration for client %s", thread_client_id);
 
    // Check for immediate shutdown
    if (atomic_load_bool(&g_should_exit)) {
      log_debug("Audio render thread stopping for client %s (g_should_exit)", thread_client_id);
      break;
    }
 
    // Check thread state before acquiring any locks to prevent use-after-destroy.
    // If we acquire locks after client is being destroyed, we'll crash with SIGSEGV
    should_continue = (((int)atomic_load_bool(&client->audio_render_thread_running) != 0) &&
                       ((int)atomic_load_bool(&client->active) != 0) && !atomic_load_bool(&client->shutting_down));
 
    if (!should_continue) {
      log_debug("Audio render thread stopping for client %s (should_continue=false)", thread_client_id);
      break;
    }
 
    if (!g_audio_mixer) {
      log_dev_every(10 * NS_PER_MS_INT, "Audio render waiting for mixer (client %s)", thread_client_id);
      // Check shutdown flag while waiting
      if (atomic_load_bool(&g_should_exit))
        break;
      platform_sleep_ns(10 * NS_PER_MS_INT);
      continue;
    }
 
    // Optimization: No mutex needed - all fields are atomic or stable.
    // client_id: atomic_t - use atomic_load for thread safety
    // active: atomic_t - use atomic_load
    // audio_queue: Assigned once at init and never changes
    const char *client_id_snapshot = client->client_id;         // Atomic read
    bool active_snapshot = atomic_load_bool(&client->active);   // Atomic read
    packet_queue_t *audio_queue_snapshot = client->audio_queue; // Stable after init
 
    // Check if client is still active after getting snapshot
    if (!active_snapshot || !audio_queue_snapshot) {
      break;
    }
 
    // Create mix excluding THIS client's audio using snapshot data
    START_TIMER("mix_%s", client_id_snapshot);
 
    // ADAPTIVE READING: Read more samples when we're behind to catch up
    // Normal: 480 samples per 10ms iteration
    // When behind: read up to 960 samples to catch up faster
    // Check source buffer levels to decide
    int samples_to_read = 480; // Default: 10ms worth
 
    // Log latency at each stage in the server pipeline
    // NOTE: Skipping mixer source latency logging due to potential use-after-free
    // when source IDs are freed during concurrent client cleanup
    if (g_audio_mixer) {
      // Log outgoing queue latency
      size_t queue_depth = packet_queue_size(audio_queue_snapshot);
      float queue_latency_ms = (float)queue_depth * 20.0f; // ~20ms per Opus packet
      log_dev_every(5 * NS_PER_MS_INT, "LATENCY: Server send queue for client %s: %.1fms (%zu packets)",
                    client_id_snapshot, queue_latency_ms, queue_depth);
    }
 
    int samples_mixed = 0;
    uint32_t client_id_hash = fnv1a_hash_string(client_id_snapshot);
    samples_mixed = mixer_process_excluding_source(g_audio_mixer, mix_buffer, samples_to_read, client_id_hash);
 
    STOP_TIMER_AND_LOG_EVERY(dev, NS_PER_SEC_INT, 5 * NS_PER_MS_INT, "mix_%s", "Mixer for client %s: took",
                             client_id_snapshot);
 
    // Debug logging every 100 iterations (disabled - can slow down audio rendering)
    // log_debug_every(LOG_RATE_SLOW, "Audio render for client %u: samples_mixed=%d", client_id_snapshot,
    // samples_mixed);
 
    // Accumulate all samples (including 0 or partial) until we have a full Opus frame
    // This maintains continuous stream without silence padding
    START_TIMER("accum_%s", client_id_snapshot);
 
    int space_available = OPUS_FRAME_SAMPLES - opus_frame_accumulated;
    int samples_to_copy = (samples_mixed <= space_available) ? samples_mixed : space_available;
 
    // Only copy if we have samples, otherwise just wait for next frame
    if (samples_to_copy > 0) {
      SAFE_MEMCPY(opus_frame_buffer + opus_frame_accumulated,
                  (OPUS_FRAME_SAMPLES - opus_frame_accumulated) * sizeof(float), mix_buffer,
                  samples_to_copy * sizeof(float));
      opus_frame_accumulated += samples_to_copy;
    }
 
    STOP_TIMER_AND_LOG_EVERY(dev, NS_PER_SEC_INT, 2 * NS_PER_MS_INT, "accum_%s", "Accumulate for client %s: took",
                             client_id_snapshot);
 
    // Only encode and send when we have accumulated a full Opus frame
    if (opus_frame_accumulated >= OPUS_FRAME_SAMPLES) {
      // A completed Opus frame is only 20 ms, while the outgoing audio queue
      // is the same ordered path that precedes video. Enforce the bound for
      // every frame: checking only once per 100 iterations lets a 200 ms
      // backlog grow past a second before the producer reacts.
      size_t queue_depth = packet_queue_size(audio_queue_snapshot);
      bool apply_backpressure = queue_depth >= 10;
      if (apply_backpressure) {
        log_warn_every(4500 * US_PER_MS_INT,
                       "Audio backpressure for client %s: queue depth %zu packets (%.1fms buffered)",
                       client_id_snapshot, queue_depth, (float)queue_depth / 50.0f * 1000.0f);
      }
 
      if (apply_backpressure) {
        // Skip this packet to let the queue drain
        // Reset accumulation buffer so fresh samples can be captured on next iteration.
        // Without this reset, we'd loop forever with stale audio and no space for new samples
        opus_frame_accumulated = 0;
        // Reduced sleep from 5.8ms to 1ms to minimize audio gaps on localhost
        platform_sleep_ns(1 * NS_PER_MS_INT);
        continue;
      }
 
      // Encode accumulated Opus frame (960 samples = 20ms @ 48kHz)
      uint8_t opus_buffer[1024]; // Max Opus frame size
 
      START_TIMER("opus_encode_%s", client_id_snapshot);
 
      int opus_size =
          opus_codec_encode(opus_encoder, opus_frame_buffer, OPUS_FRAME_SAMPLES, opus_buffer, sizeof(opus_buffer));
 
      STOP_TIMER_AND_LOG_EVERY(dev, NS_PER_SEC_INT, 10 * NS_PER_MS_INT, "opus_encode_%s",
                               "Opus encode for client %s: took", client_id_snapshot);
 
      // DEBUG: Log mix buffer and encoding results to see audio levels being sent
      {
        float peak = 0.0f, rms = 0.0f;
        for (int i = 0; i < OPUS_FRAME_SAMPLES; i++) {
          float abs_val = fabsf(opus_frame_buffer[i]);
          if (abs_val > peak)
            peak = abs_val;
          rms += opus_frame_buffer[i] * opus_frame_buffer[i];
        }
        rms = sqrtf(rms / OPUS_FRAME_SAMPLES);
        // NOTE: server_audio_frame_count is now per-thread (not static), so each client thread has its own counter
        server_audio_frame_count++;
        if (server_audio_frame_count <= 5 || server_audio_frame_count % 20 == 0) {
          // Log first 4 samples to verify they look like valid audio (not NaN/Inf/garbage)
          log_dev_every(4500 * US_PER_MS_INT,
                        "Server audio frame #%d for client %s: samples_mixed=%d, Peak=%.6f, RMS=%.6f, opus_size=%d, "
                        "first4=[%.4f,%.4f,%.4f,%.4f]",
                        server_audio_frame_count, client_id_snapshot, samples_mixed, peak, rms, opus_size,
                        opus_frame_buffer[0], opus_frame_buffer[1], opus_frame_buffer[2], opus_frame_buffer[3]);
        }
      }
 
      // Always reset accumulation buffer after attempting to encode - we've consumed these samples
      // If we don't reset, new audio samples would be dropped while stale data sits in the buffer
      opus_frame_accumulated = 0;
 
      if (opus_size <= 0) {
        log_error("Failed to encode audio to Opus for client %s: opus_size=%d", client_id_snapshot, opus_size);
      } else {
        // Wrap Opus data in proper batch packet format for client parsing:
        // Header: sample_rate (4B) + frame_duration (4B) + frame_count (4B)
        // Frame sizes: frame_count * 2B (one uint16 per frame with Opus data size)
        // Opus data: raw encoded audio
 
        // Fixed values for Opus encoding (standard 48kHz, 960 samples per frame = 20ms)
        const uint32_t sample_rate = 48000;
        const uint32_t frame_duration = 20; // milliseconds (960 samples / 48kHz = 20ms)
        const uint32_t frame_count = 1;     // one Opus frame per packet
        const uint16_t frame_size = (uint16_t)opus_size;
 
        // Calculate total packet size: 16-byte header + frame_sizes + opus data
        const size_t header_size = 16; // 4 uint32s (sample_rate, frame_duration, frame_count, reserved)
        const size_t frame_sizes_size = (size_t)frame_count * sizeof(uint16_t);
        const size_t total_packet_size = header_size + frame_sizes_size + (size_t)opus_size;
 
        // Allocate buffer for the complete packet
        uint8_t *packet_buffer = SAFE_MALLOC(total_packet_size, uint8_t *);
        if (!packet_buffer) {
          log_error("Failed to allocate Opus batch packet buffer for client %s (size=%zu)", client_id_snapshot,
                    total_packet_size);
        } else {
          // Build header in network byte order
          uint8_t *ptr = packet_buffer;
 
          // Write sample_rate (4 bytes, network byte order)
          uint32_t sr_net = htonl(sample_rate);
          memcpy(ptr, &sr_net, 4);
          ptr += 4;
 
          // Write frame_duration (4 bytes, network byte order)
          uint32_t fd_net = htonl(frame_duration);
          memcpy(ptr, &fd_net, 4);
          ptr += 4;
 
          // Write frame_count (4 bytes, network byte order)
          uint32_t fc_net = htonl(frame_count);
          memcpy(ptr, &fc_net, 4);
          ptr += 4;
 
          // Write reserved field (4 bytes, all zeros)
          uint32_t reserved = 0;
          memcpy(ptr, &reserved, 4);
          ptr += 4;
 
          // Write frame sizes (2 bytes per frame, network byte order)
          for (uint32_t i = 0; i < frame_count; i++) {
            uint16_t fs_net = htons(frame_size);
            memcpy(ptr, &fs_net, 2);
            ptr += 2;
          }
 
          // Copy Opus data
          memcpy(ptr, opus_buffer, (size_t)opus_size);
 
          // Queue the properly formatted packet
          START_TIMER("audio_queue_%s", client_id_snapshot);
 
          int result = packet_queue_enqueue(audio_queue_snapshot, PACKET_TYPE_AUDIO_OPUS_BATCH, packet_buffer,
                                            total_packet_size, 0, true);
 
          STOP_TIMER_AND_LOG_EVERY(dev, NS_PER_SEC_INT, 1 * NS_PER_MS_INT, "audio_queue_%s",
                                   "Audio queue for client %s: took", client_id_snapshot);
 
          if (result < 0) {
            log_debug("Failed to queue Opus audio for client %s", client_id_snapshot);
          } else {
            // FPS tracking - audio packet successfully queued (handles lag detection and periodic reporting)
            fps_frame_ns(&audio_fps_tracker, time_get_ns(), "audio packet queued");
          }
 
          // Free the allocated packet buffer after queuing (packet_queue_enqueue copies data)
          SAFE_FREE(packet_buffer);
        }
      }
      // NOTE: opus_frame_accumulated is already reset at line 928 after encode attempt
    }
 
    // Audio mixing rate limiting using adaptive sleep system
    // Target: 10ms intervals (100 FPS) for 480 samples @ 48kHz
    // Use queue_depth=0 and target_depth=0 for constant-rate audio processing
    audio_deadline += 10 * NS_PER_MS_INT;
    uint64_t now = time_get_ns();
    if (audio_deadline > now)
      platform_sleep_ns(audio_deadline - now);
    else if (now - audio_deadline > 100 * NS_PER_MS_INT)
      audio_deadline = now;
  }
 
#ifdef DEBUG_THREADS
  log_debug("Audio render thread stopped for client %s", thread_client_id);
#endif
 
  // Clean up Opus encoder
  if (opus_encoder) {
    opus_codec_destroy(opus_encoder);
  }
 
  // Clean up thread-local error context before exit
  asciichat_errno_destroy();
 
  return NULL;
}
 
/* ============================================================================
 * Thread Lifecycle Management Functions
 * ============================================================================
 */
 
int create_client_render_threads(server_context_t *server_ctx, client_info_t *client) {
  log_info("★★★ create_client_render_threads() CALLED for client_id=%s", client ? client->client_id : "NULL");
 
  if (!server_ctx || !client) {
    log_error("Cannot create render threads: NULL %s", !server_ctx ? "server_ctx" : "client");
    return -1;
  }
 
#ifdef DEBUG_THREADS
  log_debug("Creating render threads for client %s", client->client_id);
#endif
 
  // NOTE: Mutexes are already initialized in add_client() before any threads start
  // This prevents race conditions where receive thread tries to use uninitialized mutexes
 
  // Initialize render thread control flags
  // IMPORTANT: Set to true BEFORE creating thread to avoid race condition
  // where thread starts and immediately exits because flag is false
  atomic_store_bool(&client->video_render_thread_running, true);
  atomic_store_bool(&client->audio_render_thread_running, true);
 
  // Create video rendering thread (stop_id=2, stop after receive thread)
  char thread_name[64];
  safe_snprintf(thread_name, sizeof(thread_name), "video_render_%s", client->client_id);
  asciichat_error_t video_result;
 
  if (client->is_tcp_client) {
    video_result = tcp_server_spawn_thread(server_ctx->tcp_server, client->socket, client_video_render_thread, client,
                                           2, thread_name);
  } else {
    log_debug("THREAD_CREATE: WebRTC/WebSocket render video thread for client %s", client->client_id);
    video_result =
        asciichat_thread_create(&client->video_render_thread, thread_name, client_video_render_thread, client);
  }
 
  if (video_result != ASCIICHAT_OK) {
    // Reset flag since thread creation failed
    atomic_store_bool(&client->video_render_thread_running, false);
    // Mutexes will be destroyed by remove_client() which called us
    return -1;
  }
 
  // Create audio rendering thread (stop_id=2, same priority as video)
  safe_snprintf(thread_name, sizeof(thread_name), "audio_render_%s", client->client_id);
  asciichat_error_t audio_result;
 
  if (client->is_tcp_client) {
    audio_result = tcp_server_spawn_thread(server_ctx->tcp_server, client->socket, client_audio_render_thread, client,
                                           2, thread_name);
  } else {
    log_debug("THREAD_CREATE: WebRTC/WebSocket render audio thread for client %s", client->client_id);
    audio_result =
        asciichat_thread_create(&client->audio_render_thread, thread_name, client_audio_render_thread, client);
  }
  if (audio_result != ASCIICHAT_OK) {
    // Clean up video thread (atomic operation, no mutex needed)
    atomic_store_bool(&client->video_render_thread_running, false);
    // Reset audio flag since thread creation failed
    atomic_store_bool(&client->audio_render_thread_running, false);
    // tcp_server_stop_client_threads() will be called by remove_client()
    // to clean up the video thread we just created
    // Mutexes will be destroyed by remove_client() which called us
    return -1;
  }
 
#ifdef DEBUG_THREADS
  log_debug("Created render threads for client %s", client->client_id);
#endif
 
  return 0;
}
 
void stop_client_render_threads(client_info_t *client) {
  if (!client) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Client is NULL");
    return;
  }
 
  log_debug("Stopping render threads for client %s", client->client_id);
 
  // Signal threads to stop (atomic operations, no mutex needed)
  atomic_store_bool(&client->video_render_thread_running, false);
  atomic_store_bool(&client->audio_render_thread_running, false);
 
  // Wait for threads to finish (deterministic cleanup)
  // During shutdown, don't wait forever for threads to join
  bool is_shutting_down = atomic_load_bool(&g_should_exit);
 
  if (asciichat_thread_is_initialized(&client->video_render_thread)) {
    log_debug("Joining video render thread for client %s", client->client_id);
    int result;
    if (is_shutting_down) {
      // During shutdown, don't timeout - wait for thread to exit
      // Timeouts mask the real problem: threads that are still running
      log_debug("Shutdown mode: joining video render thread for client %s (no timeout)", client->client_id);
      result = asciichat_thread_join(&client->video_render_thread, NULL);
      if (result != 0) {
        log_warn("Video render thread for client %s failed to join during shutdown: %s", client->client_id,
                 SAFE_STRERROR(result));
      }
    } else {
      log_debug("Calling asciichat_thread_join for video thread of client %s", client->client_id);
      result = asciichat_thread_join(&client->video_render_thread, NULL);
      log_debug("asciichat_thread_join returned %d for video thread of client %s", result, client->client_id);
    }
 
    if (result == 0) {
#ifdef DEBUG_THREADS
      log_debug("Video render thread joined for client %s", client->client_id);
#endif
    } else if (result != -2) { // Don't log timeout errors again
      if (is_shutting_down) {
        log_warn("Failed to join video render thread for client %s during shutdown (continuing): %s", client->client_id,
                 SAFE_STRERROR(result));
      } else {
        log_error("Failed to join video render thread for client %s: %s", client->client_id, SAFE_STRERROR(result));
      }
    }
    if (result == 0) {
      asciichat_thread_init(&client->video_render_thread);
    }
  }
 
  if (asciichat_thread_is_initialized(&client->audio_render_thread)) {
    int result;
    if (is_shutting_down) {
      // During shutdown, don't timeout - wait for thread to exit
      // Timeouts mask the real problem: threads that are still running
      log_debug("Shutdown mode: joining audio render thread for client %s (no timeout)", client->client_id);
      result = asciichat_thread_join(&client->audio_render_thread, NULL);
      if (result != 0) {
        log_warn("Audio render thread for client %s failed to join during shutdown: %s", client->client_id,
                 SAFE_STRERROR(result));
      }
    } else {
      result = asciichat_thread_join(&client->audio_render_thread, NULL);
    }
 
    if (result == 0) {
#ifdef DEBUG_THREADS
      log_debug("Audio render thread joined for client %s", client->client_id);
#endif
    } else if (result != -2) { // Don't log timeout errors again
      if (is_shutting_down) {
        log_warn("Failed to join audio render thread for client %s during shutdown (continuing): %s", client->client_id,
                 SAFE_STRERROR(result));
      } else {
        log_error("Failed to join audio render thread for client %s: %s", client->client_id, SAFE_STRERROR(result));
      }
    }
    if (result == 0) {
      asciichat_thread_init(&client->audio_render_thread);
    }
  }
 
  // DO NOT destroy the mutex here - client.c will handle it
  // mutex_destroy(&client->client_state_mutex);
 
#ifdef DEBUG_THREADS
  log_debug("Successfully destroyed render threads for client %s", client->client_id);
#endif
}