render.c

Go to the documentation of this file.

 
#include <stdatomic.h>
#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 "common.h"
#include "options/options.h"
#include "options/rcu.h" // For RCU-based options access
#include "platform/abstraction.h"
#include "platform/init.h"
#include "network/packet_queue.h"
#include "util/time.h"
#include "audio/mixer.h"
#include "audio/audio.h"
#include "audio/opus_codec.h"
#include "util/format.h"
#include "util/fps.h"
 
// Global client manager lock for thread-safe access
extern rwlock_t g_client_manager_rwlock;
 
extern atomic_bool g_server_should_exit;
 
extern mixer_t *g_audio_mixer;
 
/* ============================================================================
 * Cross-Platform Utility Functions
 * ============================================================================
 */
 
// Removed interruptible_usleep - using regular platform_sleep_usec 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
  // CRITICAL: Use atomic_load for client_id to prevent data races
  uint32_t thread_client_id = atomic_load(&client->client_id);
  socket_t thread_socket = client->socket;
 
  log_debug("Video render thread: client_id=%u", thread_client_id);
 
  if (thread_socket == INVALID_SOCKET_VALUE) {
    log_error("Invalid socket in video render thread for client %u", thread_client_id);
    return NULL;
  }
 
  // 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 %u requested FPS: %d (has_caps=%d, desired_fps=%d)", thread_client_id, client_fps, has_caps,
              desired_fps);
  } else {
    log_debug("Client %u 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 %u render interval: %dms (%d FPS)", thread_client_id, base_frame_interval_ms, client_fps);
  struct timespec last_render_time;
  (void)clock_gettime(CLOCK_MONOTONIC, &last_render_time);
 
  // FPS tracking for video render thread
  fps_t video_fps_tracker = {0};
  fps_init(&video_fps_tracker, client_fps, "SERVER VIDEO");
 
  log_info("Video render loop STARTING for client %u", thread_client_id);
 
  bool should_continue = true;
  while (should_continue && !atomic_load(&g_server_should_exit) && !atomic_load(&client->shutting_down)) {
    log_debug_every(LOG_RATE_FAST, "Video render loop iteration for client %u", thread_client_id);
 
    // Check for immediate shutdown
    if (atomic_load(&g_server_should_exit)) {
      log_debug("Video render thread stopping for client %u (g_server_should_exit)", thread_client_id);
      break;
    }
 
    bool video_running = atomic_load(&client->video_render_thread_running);
    bool active = atomic_load(&client->active);
    bool shutting_down = atomic_load(&client->shutting_down);
 
    should_continue = video_running && active && !shutting_down;
 
    if (!should_continue) {
      log_debug("Video render thread stopping for client %u (should_continue=false: video_running=%d, active=%d, "
                "shutting_down=%d)",
                thread_client_id, video_running, active, shutting_down);
      break;
    }
 
    // Rate limiting with better shutdown responsiveness
    struct timespec current_time;
    (void)clock_gettime(CLOCK_MONOTONIC, &current_time);
 
    // Use microseconds for precision - avoid integer division precision loss
    int64_t elapsed_us = ((int64_t)(current_time.tv_sec - last_render_time.tv_sec) * 1000000LL) +
                         ((int64_t)(current_time.tv_nsec - last_render_time.tv_nsec) / 1000);
    int64_t base_frame_interval_us = (int64_t)base_frame_interval_ms * 1000;
 
    if (elapsed_us < base_frame_interval_us) {
      long sleep_us = (long)(base_frame_interval_us - elapsed_us);
      // Sleep in small chunks for reasonable shutdown response (balance performance vs responsiveness)
      const long max_sleep_chunk = 5000; // 5ms chunks for good shutdown response without destroying performance
      while (sleep_us > 0 && !atomic_load(&g_server_should_exit)) {
        long chunk = sleep_us > max_sleep_chunk ? max_sleep_chunk : sleep_us;
        platform_sleep_usec(chunk);
        sleep_us -= chunk;
 
        // Check all shutdown conditions after each tiny sleep
        if (atomic_load(&g_server_should_exit))
          break;
 
        bool still_running = atomic_load(&client->video_render_thread_running) && atomic_load(&client->active);
        if (!still_running)
          break;
      }
      // Fall through to render frame after sleeping
    }
 
    // CRITICAL: Check thread state again BEFORE acquiring locks (client might have been destroyed during sleep)
    should_continue = atomic_load(&client->video_render_thread_running) && atomic_load(&client->active) &&
                      !atomic_load(&client->shutting_down);
    if (!should_continue) {
      break;
    }
 
    // CRITICAL OPTIMIZATION: No mutex needed - all fields are atomic or stable!
    // client_id: atomic_uint - use atomic_load for thread safety
    // width/height: atomic_ushort - use atomic_load
    // active: atomic_bool - use atomic_load
    uint32_t client_id_snapshot = atomic_load(&client->client_id); // Atomic read
    unsigned short width_snapshot = atomic_load(&client->width);   // Atomic read
    unsigned short height_snapshot = atomic_load(&client->height); // Atomic read
    bool active_snapshot = atomic_load(&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();
    log_debug("Video render iteration for client %u: has_video_sources=%d", thread_client_id, has_video_sources);
 
    if (!has_video_sources) {
      // No video sources - skip frame generation but DON'T update last_render_time
      // This ensures the next iteration still maintains proper frame timing
      // DON'T continue here - let the loop update last_render_time at the bottom
      // Fall through to update last_render_time at bottom of loop
      log_debug("Skipping frame generation for client %u (no video sources)", thread_client_id);
      goto skip_frame_generation;
    }
 
    int sources_count = 0; // Track number of video sources in this frame
 
    log_debug("About to call create_mixed_ascii_frame_for_client for client %u", thread_client_id);
    char *ascii_frame = create_mixed_ascii_frame_for_client(client_id_snapshot, width_snapshot, height_snapshot, false,
                                                            &frame_size, NULL, &sources_count);
    log_debug("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("Buffering frame for client %u (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(&client->last_rendered_grid_sources, 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->capture_timestamp_us =
                (uint64_t)current_time.tv_sec * 1000000 + (uint64_t)current_time.tv_nsec / 1000;
 
            // Commit the frame (swaps buffers atomically using vfb->swap_mutex, NOT rwlock)
            video_frame_commit(vfb_snapshot);
 
            // Log occasionally for monitoring
            char pretty_size[64];
            format_bytes_pretty(frame_size, pretty_size, sizeof(pretty_size));
 
          } 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(&video_fps_tracker, &current_time, "frame rendered");
        }
      }
 
      SAFE_FREE(ascii_frame);
    } else {
      // No frame generated (probably no video sources) - this is normal, no error logging needed
      log_debug_every(LOG_RATE_NORMAL, "Per-client render: No video sources available for client %u",
                      client_id_snapshot);
    }
 
  skip_frame_generation:
    // Update last_render_time to maintain consistent frame timing
    // CRITICAL: Use the current_time captured at the START of this iteration for rate limiting.
    // This ensures we maintain the target frame rate based on when we STARTED processing,
    // not when we FINISHED. This prevents timing drift from frame generation overhead.
    last_render_time = current_time;
  }
 
#ifdef DEBUG_THREADS
  log_debug("Video render thread stopped for client %u", thread_client_id);
#endif
 
  // Clean up thread-local error context before exit
  asciichat_errno_cleanup();
 
  return NULL;
}
 
/* ============================================================================
 * Per-Client Audio Rendering Implementation
 * ============================================================================
 */
 
void *client_audio_render_thread(void *arg) {
  client_info_t *client = (client_info_t *)arg;
 
  if (!client || client->socket == INVALID_SOCKET_VALUE) {
    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
  // CRITICAL: Use atomic_load for client_id to prevent data races
  uint32_t thread_client_id = atomic_load(&client->client_id);
  char thread_display_name[64];
 
  // 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 %u (%s)", thread_client_id, thread_display_name);
#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 %u)", thread_client_id);
    return NULL;
  }
 
  // FPS tracking for audio render thread
  fps_t audio_fps_tracker = {0};
  fps_init(&audio_fps_tracker, AUDIO_RENDER_FPS, "SERVER AUDIO");
  struct timespec last_packet_send_time; // For time-based packet transmission (every 20ms)
  (void)clock_gettime(CLOCK_MONOTONIC, &last_packet_send_time);
 
  // Per-thread counters (NOT static - each thread instance gets its own)
  int mixer_debug_count = 0;
  int backpressure_check_counter = 0;
  int server_audio_frame_count = 0;
 
  bool should_continue = true;
  while (should_continue && !atomic_load(&g_server_should_exit) && !atomic_load(&client->shutting_down)) {
    // Capture loop start time for precise timing
    struct timespec loop_start_time;
    (void)clock_gettime(CLOCK_MONOTONIC, &loop_start_time);
 
    log_debug_every(LOG_RATE_SLOW, "Audio render loop iteration for client %u", thread_client_id);
 
    // Check for immediate shutdown
    if (atomic_load(&g_server_should_exit)) {
      log_debug("Audio render thread stopping for client %u (g_server_should_exit)", thread_client_id);
      break;
    }
 
    // CRITICAL: 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(&client->audio_render_thread_running) != 0) &&
                       ((int)atomic_load(&client->active) != 0) && !atomic_load(&client->shutting_down));
 
    if (!should_continue) {
      log_debug("Audio render thread stopping for client %u (should_continue=false)", thread_client_id);
      break;
    }
 
    if (!g_audio_mixer) {
      log_info_every(LOG_RATE_FAST, "Audio render waiting for mixer (client %u)", thread_client_id);
      // Check shutdown flag while waiting
      if (atomic_load(&g_server_should_exit))
        break;
      platform_sleep_usec(10000);
      continue;
    }
 
    // CRITICAL OPTIMIZATION: No mutex needed - all fields are atomic or stable!
    // client_id: atomic_uint - use atomic_load for thread safety
    // active: atomic_bool - use atomic_load
    // audio_queue: Assigned once at init and never changes
    uint32_t client_id_snapshot = atomic_load(&client->client_id); // Atomic read
    bool active_snapshot = atomic_load(&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
    struct timespec mix_start_time;
    (void)clock_gettime(CLOCK_MONOTONIC, &mix_start_time);
 
    // 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
    if (g_audio_mixer) {
      // Check source buffer latency for all sources
      for (int i = 0; i < g_audio_mixer->max_sources; i++) {
        if (g_audio_mixer->source_ids[i] != 0 && g_audio_mixer->source_ids[i] != client_id_snapshot &&
            g_audio_mixer->source_buffers[i]) {
          size_t available = audio_ring_buffer_available_read(g_audio_mixer->source_buffers[i]);
          float buffer_latency_ms = (float)available / 48.0f; // samples / (48000 / 1000)
 
          // Log source buffer latency
          log_debug_every(500000, "LATENCY: Server incoming buffer for client %u: %.1fms (%zu samples)",
                          g_audio_mixer->source_ids[i], buffer_latency_ms, available);
 
          // If buffer is getting too full, read faster to reduce latency
          if (available > 1920) {  // > 40ms buffered - read faster!
            samples_to_read = 960; // Double read to catch up (20ms worth)
            log_debug_every(LOG_RATE_DEFAULT,
                            "LATENCY WARNING: Server buffer too full for client %u: %.1fms, reading double",
                            g_audio_mixer->source_ids[i], buffer_latency_ms);
          }
        }
      }
 
      // 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_debug_every(500000, "LATENCY: Server send queue for client %u: %.1fms (%zu packets)", client_id_snapshot,
                      queue_latency_ms, queue_depth);
    }
 
    int samples_mixed = 0;
    if (GET_OPTION(no_audio_mixer)) {
      // Disable mixer.h processing: simple mixing without ducking/compression/etc
      // Just add audio from all sources except this client, no processing
      SAFE_MEMSET(mix_buffer, samples_to_read * sizeof(float), 0, samples_to_read * sizeof(float));
 
      if (g_audio_mixer) {
        int max_samples_in_frame = 0;
        // Simple mixing: just add all sources except current client
        for (int i = 0; i < g_audio_mixer->max_sources; i++) {
          if (g_audio_mixer->source_ids[i] != 0 && g_audio_mixer->source_ids[i] != client_id_snapshot &&
              g_audio_mixer->source_buffers[i]) {
            // Read from this source and add to mix buffer
            float temp_buffer[960]; // Max adaptive read size
            int samples_read =
                (int)audio_ring_buffer_read(g_audio_mixer->source_buffers[i], temp_buffer, samples_to_read);
 
            // Track the maximum samples we got from any source
            if (samples_read > max_samples_in_frame) {
              max_samples_in_frame = samples_read;
            }
 
            // Add to mix buffer
            for (int j = 0; j < samples_read; j++) {
              mix_buffer[j] += temp_buffer[j];
            }
          }
        }
        samples_mixed = max_samples_in_frame; // Only count samples we actually read
      }
 
      log_debug_every(LOG_RATE_DEFAULT,
                      "Audio mixer DISABLED (--no-audio-mixer): simple mixing, samples=%d for client %u", samples_mixed,
                      client_id_snapshot);
    } else {
      // Use adaptive sample count in normal mixer mode
      samples_mixed = mixer_process_excluding_source(g_audio_mixer, mix_buffer, samples_to_read, client_id_snapshot);
    }
 
    struct timespec mix_end_time;
    (void)clock_gettime(CLOCK_MONOTONIC, &mix_end_time);
    uint64_t mix_time_us = ((uint64_t)mix_end_time.tv_sec * 1000000 + (uint64_t)mix_end_time.tv_nsec / 1000) -
                           ((uint64_t)mix_start_time.tv_sec * 1000000 + (uint64_t)mix_start_time.tv_nsec / 1000);
 
    if (mix_time_us > 2000) { // Log if mixing takes > 2ms
      log_warn_every(LOG_RATE_DEFAULT, "Slow mixer for client %u: took %lluus (%.2fms)", client_id_snapshot,
                     mix_time_us, (float)mix_time_us / 1000.0f);
    }
 
    // 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);
 
    // DEBUG: Log samples mixed every iteration
    // NOTE: mixer_debug_count is now per-thread (not static), so each client thread has its own counter
    mixer_debug_count++;
    if (samples_mixed > 0 && (mixer_debug_count <= 3 || mixer_debug_count % 50 == 0)) {
      log_info("Server mixer iteration #%d for client %u: samples_mixed=%d, opus_frame_accumulated=%d/%d",
               mixer_debug_count, client_id_snapshot, samples_mixed, opus_frame_accumulated, OPUS_FRAME_SAMPLES);
    }
 
    // Accumulate all samples (including 0 or partial) until we have a full Opus frame
    // This maintains continuous stream without silence padding
    struct timespec accum_start = {0};
    (void)clock_gettime(CLOCK_MONOTONIC, &accum_start);
 
    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;
    }
 
    struct timespec accum_end = {0};
    (void)clock_gettime(CLOCK_MONOTONIC, &accum_end);
    uint64_t accum_time_us = ((uint64_t)accum_end.tv_sec * 1000000 + (uint64_t)accum_end.tv_nsec / 1000) -
                             ((uint64_t)accum_start.tv_sec * 1000000 + (uint64_t)accum_start.tv_nsec / 1000);
 
    if (accum_time_us > 500) {
      log_warn_every(LOG_RATE_DEFAULT, "Slow accumulate for client %u: took %lluus", client_id_snapshot, accum_time_us);
    }
 
    // Only encode and send when we have accumulated a full Opus frame
    if (opus_frame_accumulated >= OPUS_FRAME_SAMPLES) {
      // OPTIMIZATION: Don't check queue depth every iteration - it's expensive (requires lock)
      // Only check periodically every 100 iterations (~0.6s at 172 fps)
      // NOTE: backpressure_check_counter is now per-thread (not static), so each client thread has its own counter
      bool apply_backpressure = false;
 
      if (++backpressure_check_counter >= 100) {
        backpressure_check_counter = 0;
        size_t queue_depth = packet_queue_size(audio_queue_snapshot);
        // Opus frames are produced at ~50 FPS (20ms each), so 50 packets = 1 second
        // Keep latency bounded to ~1s max in the send queue
        apply_backpressure = (queue_depth > 50); // > 50 packets = ~1s buffered at 50 FPS
 
        if (apply_backpressure) {
          log_warn("Audio backpressure for client %u: queue depth %zu packets (%.1fs buffered)", client_id_snapshot,
                   queue_depth, (float)queue_depth / 50.0f);
        }
      }
 
      if (apply_backpressure) {
        // Skip this packet to let the queue drain
        // CRITICAL: 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;
        platform_sleep_usec(5800);
        continue;
      }
 
      // Encode accumulated Opus frame (960 samples = 20ms @ 48kHz)
      uint8_t opus_buffer[1024]; // Max Opus frame size
 
      struct timespec opus_start_time;
      (void)clock_gettime(CLOCK_MONOTONIC, &opus_start_time);
 
      int opus_size =
          opus_codec_encode(opus_encoder, opus_frame_buffer, OPUS_FRAME_SAMPLES, opus_buffer, sizeof(opus_buffer));
 
      struct timespec opus_end_time;
      (void)clock_gettime(CLOCK_MONOTONIC, &opus_end_time);
      uint64_t opus_time_us = ((uint64_t)opus_end_time.tv_sec * 1000000 + (uint64_t)opus_end_time.tv_nsec / 1000) -
                              ((uint64_t)opus_start_time.tv_sec * 1000000 + (uint64_t)opus_start_time.tv_nsec / 1000);
 
      if (opus_time_us > 2000) { // Log if encoding takes > 2ms
        log_warn_every(LOG_RATE_DEFAULT, "Slow Opus encode for client %u: took %lluus (%.2fms), size=%d",
                       client_id_snapshot, opus_time_us, (float)opus_time_us / 1000.0f, opus_size);
      }
 
      // 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_info("Server audio frame #%d for client %u: 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 %u: opus_size=%d", client_id_snapshot, opus_size);
      } else {
        // Queue Opus-encoded audio for this specific client
        struct timespec queue_start = {0};
        (void)clock_gettime(CLOCK_MONOTONIC, &queue_start);
 
        int result =
            packet_queue_enqueue(audio_queue_snapshot, PACKET_TYPE_AUDIO_OPUS, opus_buffer, (size_t)opus_size, 0, true);
 
        struct timespec queue_end = {0};
        (void)clock_gettime(CLOCK_MONOTONIC, &queue_end);
        uint64_t queue_time_us = ((uint64_t)queue_end.tv_sec * 1000000 + (uint64_t)queue_end.tv_nsec / 1000) -
                                 ((uint64_t)queue_start.tv_sec * 1000000 + (uint64_t)queue_start.tv_nsec / 1000);
 
        if (queue_time_us > 500) {
          log_warn_every(LOG_RATE_DEFAULT, "Slow queue for client %u: took %lluus", client_id_snapshot, queue_time_us);
        }
 
        if (result < 0) {
          log_debug("Failed to queue Opus audio for client %u", client_id_snapshot);
        } else {
          // FPS tracking - audio packet successfully queued (handles lag detection and periodic reporting)
          struct timespec current_time;
          (void)clock_gettime(CLOCK_MONOTONIC, &current_time);
          fps_frame(&audio_fps_tracker, &current_time, "audio packet queued");
        }
      }
      // NOTE: opus_frame_accumulated is already reset at line 928 after encode attempt
    }
 
    // Audio mixing rate - 10ms to match 48kHz sample rate (480 samples per iteration)
    // Use ABSOLUTE timing to prevent drift from chunked sleep overhead
    // Previous approach slept in 1ms chunks causing ~20% timing drift (12ms instead of 10ms)
    // which drained client buffers faster than they were filled.
    struct timespec loop_end_time;
    (void)clock_gettime(CLOCK_MONOTONIC, &loop_end_time);
 
    uint64_t loop_elapsed_us = ((uint64_t)loop_end_time.tv_sec * 1000000 + (uint64_t)loop_end_time.tv_nsec / 1000) -
                               ((uint64_t)loop_start_time.tv_sec * 1000000 + (uint64_t)loop_start_time.tv_nsec / 1000);
 
    // Target loop time: 10ms to match the audio sample rate
    // We read 480 samples per iteration at 48kHz: 480 / 48000 = 0.01s = 10ms
    const uint64_t target_loop_us = 10000; // 10ms for exactly 48kHz rate
 
    if (loop_elapsed_us >= target_loop_us) {
      // Processing took longer than target - skip sleep but warn
      log_warn_every(LOG_RATE_FAST,
                     "Audio processing took %lluus (%.1fms) - exceeds target %lluus (%.1fms) for client %u",
                     loop_elapsed_us, (float)loop_elapsed_us / 1000.0f, target_loop_us, (float)target_loop_us / 1000.0f,
                     thread_client_id);
    } else {
      // Sleep for remaining time in ONE call to avoid chunked sleep overhead
      // Check shutdown first, then do one accurate sleep
      if (!atomic_load(&g_server_should_exit) && atomic_load(&client->audio_render_thread_running)) {
        long remaining_sleep_us = (long)(target_loop_us - loop_elapsed_us);
        platform_sleep_usec(remaining_sleep_us);
      }
    }
  }
 
#ifdef DEBUG_THREADS
  log_debug("Audio render thread stopped for client %u", 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_cleanup();
 
  return NULL;
}
 
/* ============================================================================
 * Thread Lifecycle Management Functions
 * ============================================================================
 */
 
int create_client_render_threads(server_context_t *server_ctx, client_info_t *client) {
  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 %u", 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(&client->video_render_thread_running, true);
  atomic_store(&client->audio_render_thread_running, true);
 
  // Create video rendering thread (stop_id=2, stop after receive thread)
  char thread_name[64];
  snprintf(thread_name, sizeof(thread_name), "video_render_%u", client->client_id);
  asciichat_error_t video_result = tcp_server_spawn_thread(server_ctx->tcp_server, client->socket,
                                                           client_video_render_thread, client, 2, thread_name);
  if (video_result != ASCIICHAT_OK) {
    // Reset flag since thread creation failed
    atomic_store(&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)
  snprintf(thread_name, sizeof(thread_name), "audio_render_%u", client->client_id);
  asciichat_error_t audio_result = tcp_server_spawn_thread(server_ctx->tcp_server, client->socket,
                                                           client_audio_render_thread, client, 2, thread_name);
  if (audio_result != ASCIICHAT_OK) {
    // Clean up video thread (atomic operation, no mutex needed)
    atomic_store(&client->video_render_thread_running, false);
    // Reset audio flag since thread creation failed
    atomic_store(&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 %u", 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 %u", client->client_id);
 
  // Signal threads to stop (atomic operations, no mutex needed)
  atomic_store(&client->video_render_thread_running, false);
  atomic_store(&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(&g_server_should_exit);
 
  if (asciichat_thread_is_initialized(&client->video_render_thread)) {
    log_debug("Joining video render thread for client %u", 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 %u (no timeout)", client->client_id);
      result = asciichat_thread_join(&client->video_render_thread, NULL);
      if (result != 0) {
        log_warn("Video render thread for client %u failed to join during shutdown: %s", client->client_id,
                 SAFE_STRERROR(result));
      }
    } else {
      log_debug("Calling asciichat_thread_join for video thread of client %u", client->client_id);
      result = asciichat_thread_join(&client->video_render_thread, NULL);
      log_debug("asciichat_thread_join returned %d for video thread of client %u", result, client->client_id);
    }
 
    if (result == 0) {
#ifdef DEBUG_THREADS
      log_debug("Video render thread joined for client %u", 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 %u during shutdown (continuing): %s", client->client_id,
                 SAFE_STRERROR(result));
      } else {
        log_error("Failed to join video render thread for client %u: %s", client->client_id, SAFE_STRERROR(result));
      }
    }
    // Clear thread handle safely using platform abstraction
    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 %u (no timeout)", client->client_id);
      result = asciichat_thread_join(&client->audio_render_thread, NULL);
      if (result != 0) {
        log_warn("Audio render thread for client %u 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 %u", 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 %u during shutdown (continuing): %s", client->client_id,
                 SAFE_STRERROR(result));
      } else {
        log_error("Failed to join audio render thread for client %u: %s", client->client_id, SAFE_STRERROR(result));
      }
    }
    // Clear thread handle safely using platform abstraction
    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 %u", client->client_id);
#endif
}