src/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/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/util/time.h>
#include <ascii-chat/audio/mixer.h>
#include <ascii-chat/audio/audio.h>
#include <ascii-chat/audio/opus_codec.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
  // 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;
  bool is_webrtc = (thread_socket == INVALID_SOCKET_VALUE);
  (void)is_webrtc; // May be unused in release builds
 
  log_debug("Video render thread: client_id=%u, webrtc=%d", thread_client_id, is_webrtc);
 
  // 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);
 
  // FPS tracking for video render thread
  fps_t video_fps_tracker = {0};
  fps_init(&video_fps_tracker, client_fps, "SERVER VIDEO");
 
  // Adaptive sleep for frame rate limiting
  adaptive_sleep_state_t sleep_state = {0};
  adaptive_sleep_config_t config = {
      .baseline_sleep_ns = (uint64_t)(NS_PER_SEC_INT / client_fps), // Dynamic FPS (typically 16.67ms for 60 FPS)
      .min_speed_multiplier = 1.0,                                  // Constant rate (no slowdown)
      .max_speed_multiplier = 1.0,                                  // Constant rate (no speedup)
      .speedup_rate = 0.0,                                          // No adaptive behavior (constant FPS)
      .slowdown_rate = 0.0                                          // No adaptive behavior (constant FPS)
  };
  adaptive_sleep_init(&sleep_state, &config);
 
  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_dev_every(10 * NS_PER_MS_INT, "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;
    }
 
    // Frame rate limiting using adaptive sleep system
    // Use queue_depth=0 and target_depth=0 for constant-rate renderer (no backlog management)
    adaptive_sleep_do(&sleep_state, 0, 0);
 
    // 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(&client->video_render_thread_running) && atomic_load(&client->active) &&
                      !atomic_load(&client->shutting_down);
    if (!should_continue) {
      break;
    }
 
    // 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();
 
    // DIAGNOSTIC: Track when video sources become available
    static bool last_has_sources = false;
    if (has_video_sources != last_has_sources) {
      log_warn("DIAGNOSTIC: Client %u 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 %u: has_video_sources=%d, width=%u, height=%u", thread_client_id,
                    has_video_sources, width_snapshot, height_snapshot);
 
    // Skip frame generation if client dimensions are not yet received (width=0 or height=0)
    if (width_snapshot == 0 || height_snapshot == 0) {
      log_dev_every(5 * NS_PER_MS_INT,
                    "Skipping frame generation for client %u: dimensions not yet received (width=%u, height=%u)",
                    thread_client_id, width_snapshot, height_snapshot);
      continue;
    }
 
    if (has_video_sources) {
      int sources_count = 0; // Track number of video sources in this frame
 
      // DIAGNOSTIC: Track every frame generation attempt
      // Use per-client static variables to track frequency independently
      static uint32_t frame_gen_count = 0;
      static uint64_t frame_gen_start_time = 0;
 
      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_warn("DIAGNOSTIC: Client %u LOOP running at %.1f FPS (120 iterations in %.2fs)", thread_client_id, gen_fps,
                 elapsed_ns / (double)NS_PER_SEC_INT);
      }
 
      log_dev_every(5 * NS_PER_MS_INT,
                    "About to call create_mixed_ascii_frame_for_client for client %u with dims %ux%u", thread_client_id,
                    width_snapshot, height_snapshot);
      char *ascii_frame = create_mixed_ascii_frame_for_client(client_id_snapshot, width_snapshot, height_snapshot,
                                                              false, &frame_size, NULL, &sources_count);
 
      // DEBUG: Log frame generation details
      static uint32_t last_frame_hash = -1; // Initialize to -1 so first frame is always new
      uint32_t current_frame_hash = 0;
      bool frame_is_new = false;
      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_info("RENDER_FRAME CHANGE: Client %u 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;
          frame_is_new = true;
        } else {
          log_dev_every(25000, "RENDER_FRAME DUPLICATE: Client %u frame #%zu sources=%d hash=0x%08x (no change)",
                        thread_client_id, frame_size, sources_count, current_frame_hash);
          frame_is_new = false;
        }
      }
 
      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 %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_ns = current_time_ns;
 
              // Only commit the frame if it's actually NEW (different from last committed frame)
              // This prevents sending duplicate frames and improves client-side FPS tracking
              if (frame_is_new) {
                uint64_t commit_start_ns = time_get_ns();
                // Commit the frame (swaps buffers atomically using vfb->swap_mutex, NOT rwlock)
                video_frame_commit(vfb_snapshot);
                uint64_t commit_end_ns = time_get_ns();
                char commit_duration_str[32];
                format_duration_ns((double)(commit_end_ns - commit_start_ns), commit_duration_str,
                                   sizeof(commit_duration_str));
 
                static uint32_t commits_count = 0;
                static uint64_t commits_start_time = 0;
                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_warn("DIAGNOSTIC: Client %u UNIQUE frames being sent at %.1f FPS (10 commits counted)",
                           thread_client_id, commit_fps);
                }
 
                log_info("[FRAME_COMMIT_TIMING] Client %u frame commit took %s (hash=0x%08x)", thread_client_id,
                         commit_duration_str, current_frame_hash);
              } else {
                // Discard duplicate frame by not committing (back buffer is safe to reuse)
                log_dev_every(25000, "Skipping commit for duplicate frame for client %u (hash=0x%08x)",
                              thread_client_id, current_frame_hash);
              }
 
              // Log occasionally for monitoring
              char pretty_size[64];
              format_bytes_pretty(frame_size, pretty_size, sizeof(pretty_size));
 
              // Compute hash of ASCII frame to detect duplicates
              uint32_t ascii_hash = 0;
              for (size_t i = 0; i < frame_size && i < 1000; i++) {
                ascii_hash = (uint32_t)((((uint64_t)ascii_hash << 5) - ascii_hash) + (unsigned char)ascii_frame[i]);
              }
              log_dev_every(5 * NS_PER_MS_INT, "Client %u: Rendered ASCII frame size=%s hash=0x%08x sources=%d",
                            thread_client_id, pretty_size, ascii_hash, sources_count);
 
            } 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");
          }
        }
 
        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 %u",
                      client_id_snapshot);
      }
    } else {
      // No video sources - skip frame generation but DON'T update last_render_time
      // This ensures the next iteration still maintains proper frame timing
      log_debug("Skipping frame generation for client %u (no video sources)", thread_client_id);
    }
  }
 
#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_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
  // 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];
  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 %u (%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 %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");
 
  // Adaptive sleep for audio rate limiting at 100 FPS (10ms intervals, 480 samples @ 48kHz)
  adaptive_sleep_state_t audio_sleep_state = {0};
  adaptive_sleep_config_t audio_config = {
      .baseline_sleep_ns = 10 * NS_PER_MS_INT, // 10ms = 100 FPS (480 samples @ 48kHz)
      .min_speed_multiplier = 1.0,             // Constant rate (no slowdown)
      .max_speed_multiplier = 1.0,             // Constant rate (no speedup)
      .speedup_rate = 0.0,                     // No adaptive behavior (constant rate)
      .slowdown_rate = 0.0                     // No adaptive behavior (constant rate)
  };
  adaptive_sleep_init(&audio_sleep_state, &audio_config);
 
  // 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)) {
    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;
    }
 
    // 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_dev_every(10 * NS_PER_MS_INT, "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_ns(10 * NS_PER_MS_INT);
      continue;
    }
 
    // 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
    START_TIMER("mix_%u", 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
    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_dev_every(5 * NS_PER_MS_INT, "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_dev_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_dev_every(5 * NS_PER_MS_INT, "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);
    }
 
    STOP_TIMER_AND_LOG_EVERY(dev, NS_PER_SEC_INT, 5 * NS_PER_MS_INT, "mix_%u", "Mixer for client %u: 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);
 
    // 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++;
    log_dev_every(4500 * US_PER_MS_INT,
                  "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
    START_TIMER("accum_%u", 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_%u", "Accumulate for client %u: took",
                             client_id_snapshot);
 
    // 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_every(4500 * US_PER_MS_INT,
                         "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
        // 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_ns(5800 * NS_PER_US_INT);
        continue;
      }
 
      // Encode accumulated Opus frame (960 samples = 20ms @ 48kHz)
      uint8_t opus_buffer[1024]; // Max Opus frame size
 
      START_TIMER("opus_encode_%u", 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_%u",
                               "Opus encode for client %u: 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 %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
        START_TIMER("audio_queue_%u", client_id_snapshot);
 
        int result = packet_queue_enqueue(audio_queue_snapshot, PACKET_TYPE_AUDIO_OPUS_BATCH, opus_buffer,
                                          (size_t)opus_size, 0, true);
 
        STOP_TIMER_AND_LOG_EVERY(dev, NS_PER_SEC_INT, 1 * NS_PER_MS_INT, "audio_queue_%u",
                                 "Audio queue for client %u: took", client_id_snapshot);
 
        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)
          fps_frame_ns(&audio_fps_tracker, time_get_ns(), "audio packet queued");
        }
      }
      // 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
    adaptive_sleep_do(&audio_sleep_state, 0, 0);
  }
 
#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_destroy();
 
  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];
  safe_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)
  safe_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
}