client.c File Reference

👥 Per-client lifecycle manager: threading coordination, state management, and client lifecycle orchestration More...

Go to the source code of this file.

Data Structures
struct	client_dispatch_entry_t
	Hash table entry for client packet dispatch. More...

Macros
#define	DEBUG_NETWORK   1
#define	DEBUG_THREADS   1
#define	DEBUG_MEMORY   1
#define	CLIENT_DISPATCH_HASH_SIZE   32
#define	CLIENT_DISPATCH_HANDLER_COUNT   13
#define	CLIENT_DISPATCH_HASH(type)   ((type) % CLIENT_DISPATCH_HASH_SIZE)
#define	MAX_AUDIO_BATCH   3
#define	AUDIO_OPUS_BATCH_HEADER_SIZE   16
#define	AUDIO_OPUS_FRAME_MAX_BYTES   1024

Typedefs
typedef void(*	client_packet_handler_t) (client_info_t *client, const void *data, size_t len)

Functions
void *	client_send_thread_func (void *arg)
	Client packet send thread.
void *	client_dispatch_thread (void *arg)
	Async dispatch thread for WebRTC clients.
void	broadcast_server_state_to_all_clients (void)
	Notify all clients of state changes.
client_info_t *	find_client_by_id (const char *client_id)
	Fast O(1) client lookup by ID using hash table.
client_info_t *	find_client_by_socket (socket_t socket)
	Find client by socket descriptor using linear search.
client_info_t *	add_client (server_context_t *server_ctx, socket_t socket, const char *client_ip, int port)
client_info_t *	add_webrtc_client (server_context_t *server_ctx, acip_transport_t *transport, const char *client_ip, bool start_threads)
	Register a WebRTC client with the server.
int	remove_client (server_context_t *server_ctx, const char *client_id)
void *	client_receive_thread (void *arg)
int	start_webrtc_client_threads (server_context_t *server_ctx, const char *client_id)
	Start threads for a WebRTC client after crypto initialization.
void	stop_client_threads (client_info_t *client)
void	cleanup_client_media_buffers (client_info_t *client)
void	cleanup_client_packet_queues (client_info_t *client)
int	process_encrypted_packet (client_info_t *client, packet_type_t *type, void **data, size_t *len, uint32_t *sender_id)
void	process_decrypted_packet (client_info_t *client, packet_type_t type, void *data, size_t len)

Variables
client_manager_t	g_client_manager
	Global client manager singleton - central coordination point.
rwlock_t	g_client_manager_rwlock = {0}
	Reader-writer lock protecting the global client manager.
bool	g_client_manager_rwlock_initialized = false

Detailed Description

👥 Per-client lifecycle manager: threading coordination, state management, and client lifecycle orchestration

1. Client connection establishment and initializationinitializationPer-client thread creation and management (receive, send, render))Client state management with thread-safe access patternspatternsClient disconnection handling and resource cleanupcleanupHash table management for O(1) client lookupslookupsIntegration point between main.c and other modules THREADING ARCHITECTURE PER CLIENT:

Each connected client spawns multiple dedicated threads:

• Receive Thread: Handles incoming packets from client (calls protocol.c functions)
• Send Thread: Manages outgoing packet delivery using packet queues
• Video Render Thread: Generates ASCII frames at 60fps (render.c)
• Audio Render Thread: Mixes audio streams at 172fps (render.c)

This per-client threading model provides several advantages:

• Linear performance scaling (no shared bottlenecks)
• Fault isolation (one client's issues don't affect others)
• Simplified synchronization (each client owns its resources)
• Real-time performance guarantees per client

DATA STRUCTURES AND SYNCHRONIZATION:

The module manages two primary data structures:

1. client_manager_t (global singleton):
   • Array of client_info_t structs (backing storage)
   • Hash table for O(1) client_id -> client_info_t lookups
   • Protected by reader-writer lock (g_client_manager_rwlock)
   • Allows concurrent reads, exclusive writes
2. client_info_t (per-client state):
   • Network connection details and capabilities
   • Thread handles and synchronization primitives
   • Media buffers (video/audio) and packet queues
   • Terminal capabilities and rendering preferences
   • Protected by per-client mutex (client_state_mutex)

Synchronization Patterns:

LOCK ORDERING PROTOCOL (prevents deadlocks):

1. Always acquire g_client_manager_rwlock FIRST
2. Then acquire per-client mutexes if needed
3. Release in reverse order

SNAPSHOT PATTERN (reduces lock contention):

1. Acquire mutex
2. Copy needed state variables to local copies
3. Release mutex immediately
4. Process using local copies without locks

INTEGRATION WITH OTHER MODULES:

• main.c: Calls add_client() and remove_client() from main loop
• protocol.c: Functions called by client receive threads for packet processing
• render.c: Render thread functions created per client
• stream.c: Stream generation functions called by render threads
• stats.c: Accesses client data for performance monitoring

THREAD LIFECYCLE MANAGEMENT:

Thread creation order (in add_client()):

1. Initialize client data structures and mutexes
2. Create send thread (for outgoing packet delivery)
3. Create receive thread (for incoming packet processing)
4. Create render threads (video + audio generation)

Thread termination order (in remove_client()):

1. Set shutdown flags (causes threads to exit main loops)
2. Join send thread (cleanest exit, no blocking I/O)
3. Join receive thread (may be blocked on network I/O)
4. Join render threads (computational work, clean exit)
5. Clean up resources (queues, buffers, mutexes)

WHY THIS MODULAR DESIGN:

The original server.c contained all client management code inline, making it:

• Hard to understand the client lifecycle
• Difficult to modify threading behavior
• Impossible to isolate client-related bugs
• Challenging to add new client features

This modular approach provides:

• Clear separation of client vs. server concerns
• Easier testing of client management logic
• Better code reuse and maintenance
• Future extensibility for new client types

Author

Zachary Fogg me@zf.nosp@m.o.gg

Date

September 2025

Version

2.0 (Post-Modularization)

See also

main.c For overall server architecture

render.c For per-client rendering implementation

protocol.c For client Packet Types processing

Definition in file src/server/client.c.

Macro Definition Documentation

AUDIO_OPUS_BATCH_HEADER_SIZE

#define AUDIO_OPUS_BATCH_HEADER_SIZE   16

Definition at line 2125 of file src/server/client.c.

AUDIO_OPUS_FRAME_MAX_BYTES

#define AUDIO_OPUS_FRAME_MAX_BYTES   1024

Definition at line 2126 of file src/server/client.c.

CLIENT_DISPATCH_HANDLER_COUNT

#define CLIENT_DISPATCH_HANDLER_COUNT   13

Definition at line 174 of file src/server/client.c.

CLIENT_DISPATCH_HASH

#define CLIENT_DISPATCH_HASH

(

type

)

((type) % CLIENT_DISPATCH_HASH_SIZE)

Definition at line 184 of file src/server/client.c.

CLIENT_DISPATCH_HASH_SIZE

#define CLIENT_DISPATCH_HASH_SIZE   32

Definition at line 173 of file src/server/client.c.

DEBUG_MEMORY

#define DEBUG_MEMORY   1

Definition at line 165 of file src/server/client.c.

DEBUG_NETWORK

#define DEBUG_NETWORK   1

Definition at line 163 of file src/server/client.c.

DEBUG_THREADS

#define DEBUG_THREADS   1

Definition at line 164 of file src/server/client.c.

MAX_AUDIO_BATCH

#define MAX_AUDIO_BATCH   3

Definition at line 2124 of file src/server/client.c.

Typedef Documentation

client_packet_handler_t

typedef void(* client_packet_handler_t) (client_info_t *client, const void *data, size_t len)

Definition at line 171 of file src/server/client.c.

Function Documentation

add_client()

client_info_t * add_client	(	server_context_t *	server_ctx,
		socket_t	socket,
		const char *	client_ip,
		int	port
	)

Definition at line 656 of file src/server/client.c.

                                                                                                          {
  log_info("[TCP_ADD_CLIENT] ENTER: client_ip=%s, port=%d", client_ip, port);
  // Find empty slot WITHOUT holding the global lock
  // We'll re-verify under lock after allocations complete
  int slot = -1;
  int existing_count = 0;
  for (int i = 0; i < MAX_CLIENTS; i++) {
    if (slot == -1 && g_client_manager.clients[i].client_id[0] == '\0') {
      slot = i; // Take first available slot
    }
    if (g_client_manager.clients[i].client_id[0] != '\0' && atomic_load_bool(&g_client_manager.clients[i].active)) {
      existing_count++;
    }
  }
 
  // Quick pre-check before expensive allocations
  log_info("[TCP_DBG] SLOT_CHECK: existing_count=%d, max_clients=%d, slot=%d", existing_count, GET_OPTION(max_clients),
           slot);
  if (existing_count >= GET_OPTION(max_clients) || slot == -1) {
    const char *reject_msg = "SERVER_FULL: Maximum client limit reached\n";
    ssize_t send_result = socket_send(socket, reject_msg, strlen(reject_msg), 0);
    if (send_result < 0) {
      log_warn("Failed to send rejection message to client: %s", SAFE_STRERROR(errno));
    }
    return NULL;
  }
 
  // NOW acquire the lock early for generating unique client name
  log_info("[TCP_DBG] LOCK_ACQUIRE_START");
  rwlock_wrlock(&g_client_manager_rwlock);
  log_info("[TCP_DBG] LOCK_ACQUIRED");
 
  // Generate unique client ID (noun only, without counter or port)
  char new_client_id[MAX_CLIENT_ID_LEN];
  log_info("[TCP_DBG] GENERATE_ID_START");
  if (generate_client_id(new_client_id, sizeof(new_client_id)) != 0) {
    rwlock_wrunlock(&g_client_manager_rwlock);
    log_error("Failed to generate unique client ID");
    return NULL;
  }
  log_info("[TCP_DBG] GENERATE_ID_DONE: id=%s", new_client_id);
 
  log_info("[TCP_DBG] LOCK_RELEASE_START");
  rwlock_wrunlock(&g_client_manager_rwlock);
  log_info("[TCP_DBG] LOCK_RELEASED");
 
  // DO EXPENSIVE ALLOCATIONS OUTSIDE THE LOCK
  // This prevents blocking frame processing during client initialization
  log_info("[TCP_DBG] VIDEO_CREATE_START: About to create video frame buffer for client_id=%s", new_client_id);
  video_frame_buffer_t *incoming_video_buffer = video_frame_buffer_create(new_client_id);
  log_info("[TCP_DBG] VIDEO_CREATE_DONE: Video buffer creation returned %p", (void *)incoming_video_buffer);
  if (!incoming_video_buffer) {
    SET_ERRNO(ERROR_MEMORY, "Failed to create video buffer for client %s", new_client_id);
    log_error("Failed to create video buffer for client %s", new_client_id);
    return NULL;
  }
 
  log_info("[TCP_DBG] VIDEO_DONE: About to create audio buffer");
  audio_ring_buffer_t *incoming_audio_buffer = audio_ring_buffer_create_for_capture();
  log_info("[TCP_DBG] AUDIO_DONE: Audio buffer created");
  if (!incoming_audio_buffer) {
    SET_ERRNO(ERROR_MEMORY, "Failed to create audio buffer for client %s", new_client_id);
    log_error("Failed to create audio buffer for client %s", new_client_id);
    video_frame_buffer_destroy(incoming_video_buffer);
    return NULL;
  }
 
  log_info("[TCP_DBG] AUDIO_QUEUE_START: Creating audio queue");
  // Keep the per-client queue allocation independent. Registering a 1000-node
  // pool for every client can block a second client while the first client's
  // debug registrations are being retired during disconnect.
  packet_queue_t *audio_queue = packet_queue_create(500);
  log_info("[TCP_DBG] AUDIO_QUEUE_DONE");
  if (!audio_queue) {
    LOG_ERRNO_IF_SET("Failed to create audio queue for client");
    audio_ring_buffer_destroy(incoming_audio_buffer);
    video_frame_buffer_destroy(incoming_video_buffer);
    return NULL;
  }
 
  log_info("[TCP_DBG] OUTGOING_VID_START: Creating outgoing video buffer");
  video_frame_buffer_t *outgoing_video_buffer = video_frame_buffer_create(new_client_id);
  log_info("[TCP_DBG] OUTGOING_VID_DONE");
  if (!outgoing_video_buffer) {
    LOG_ERRNO_IF_SET("Failed to create outgoing video buffer for client");
    packet_queue_destroy(audio_queue);
    audio_ring_buffer_destroy(incoming_audio_buffer);
    video_frame_buffer_destroy(incoming_video_buffer);
    return NULL;
  }
 
  void *send_buffer = SAFE_MALLOC_ALIGNED(MAX_FRAME_BUFFER_SIZE, 64, void *);
  if (!send_buffer) {
    log_error("Failed to allocate send buffer for client %s", new_client_id);
    video_frame_buffer_destroy(outgoing_video_buffer);
    packet_queue_destroy(audio_queue);
    audio_ring_buffer_destroy(incoming_audio_buffer);
    video_frame_buffer_destroy(incoming_video_buffer);
    return NULL;
  }
 
  // NOW acquire the lock for the critical section: slot assignment + registration
  log_info("[TCP_DBG] LOCK_START: About to acquire client manager write lock");
  rwlock_wrlock(&g_client_manager_rwlock);
  log_info("[TCP_DBG] LOCK_ACQUIRED: Client manager lock acquired");
  log_info("[TCP_DBG] SLOT_RECHECK: About to check if slot is still available");
 
  // Re-check slot availability under lock (another thread might have taken it)
  if (g_client_manager.clients[slot].client_id[0] != '\0') {
    rwlock_wrunlock(&g_client_manager_rwlock);
    SAFE_FREE(send_buffer);
    video_frame_buffer_destroy(outgoing_video_buffer);
    packet_queue_destroy(audio_queue);
    audio_ring_buffer_destroy(incoming_audio_buffer);
    video_frame_buffer_destroy(incoming_video_buffer);
 
    const char *reject_msg = "SERVER_FULL: Slot reassigned, try again\n";
    socket_send(socket, reject_msg, strlen(reject_msg), 0);
    return NULL;
  }
 
  // Now we have exclusive access to the slot - do the actual registration
  client_info_t *client = &g_client_manager.clients[slot];
  memset(client, 0, sizeof(client_info_t));
 
  // Initialize codec capabilities to zero (no codecs supported until CLIENT_CAPABILITIES received)
  client->codec_capabilities_video = 0;
  client->codec_capabilities_audio = 0;
 
  client->socket = socket;
  client->is_tcp_client = true;
  SAFE_STRNCPY(client->client_id, new_client_id, sizeof(client->client_id) - 1);
  SAFE_STRNCPY(client->client_ip, client_ip, sizeof(client->client_ip) - 1);
  client->port = port;
  atomic_store_bool(&client->active, true);
  client->server_ctx = server_ctx; // Store server context for cleanup
  atomic_store_bool(&client->shutting_down, false);
  atomic_store_u64(&client->last_rendered_grid_sources, 0);
  atomic_store_u64(&client->last_sent_grid_sources, 0);
  client->connected_at = time(NULL);
 
  memset(&client->crypto_handshake_ctx, 0, sizeof(client->crypto_handshake_ctx));
  client->crypto_initialized = false;
  client->transport_encrypted = false;
 
  client->pending_packet_type = 0;
  client->pending_packet_payload = NULL;
  client->pending_packet_length = 0;
 
  // Assign pre-allocated buffers
  client->incoming_video_buffer = incoming_video_buffer;
  client->incoming_audio_buffer = incoming_audio_buffer;
  client->audio_queue = audio_queue;
  client->outgoing_video_buffer = outgoing_video_buffer;
  client->send_buffer = send_buffer;
  client->send_buffer_size = MAX_FRAME_BUFFER_SIZE;
 
  // Generate unique noun-based client name with transport type and port
  // Note: We're holding the write lock here, so it's safe to iterate existing clients
  log_info("[TCP_DBG] BEFORE_GENERATE_CLIENT_NAME");
  if (generate_client_name(client->display_name, sizeof(client->display_name), g_client_manager.clients_by_id, port,
                           true /* is_tcp */) != 0) {
    // Fallback to numeric name if generation fails
    safe_snprintf(client->display_name, sizeof(client->display_name), "client_%u (tcp:%d)", new_client_id, port);
  }
  log_info("[TCP_DBG] AFTER_GENERATE_CLIENT_NAME: display_name=%s", client->display_name);
 
  // Register client with named debug system using the generated name
  (void)NAMED_REGISTER_CLIENT(client, client->display_name, NULL);
 
  log_info("Added new client %s from %s:%d (socket=%d, slot=%d)", new_client_id, client_ip, port, socket, slot);
  log_debug("Client slot assigned: client_id=%s assigned to slot %d, socket=%d", new_client_id, slot, socket);
 
  // Register socket with tcp_server
  asciichat_error_t reg_result = tcp_server_add_client(server_ctx->tcp_server, socket, client);
  if (reg_result != ASCIICHAT_OK) {
    SET_ERRNO(ERROR_INTERNAL, "Failed to register client socket with tcp_server");
    log_error("Failed to register client %s socket with tcp_server", new_client_id);
    // Don't unlock here - error_cleanup will do it
    goto error_cleanup;
  }
 
  g_client_manager.client_count++;
  log_debug("Client count updated: now %d clients (added client_id=%s to slot %d)", g_client_manager.client_count,
            new_client_id, slot);
 
  HASH_ADD_STR(g_client_manager.clients_by_id, client_id, client);
  log_debug("Added client %s to uthash table", new_client_id);
 
  rwlock_wrunlock(&g_client_manager_rwlock);
 
  // Configure socket OUTSIDE lock
  configure_client_socket(socket, new_client_id);
 
  // Initialize mutexes OUTSIDE lock BEFORE registering them
  // register_client_info_atomics needs these mutexes to be initialized first
  if (mutex_init(&client->client_state_mutex, "client_state") != 0) {
    log_error("Failed to initialize client state mutex for client %s", new_client_id);
    remove_client(server_ctx, new_client_id);
    return NULL;
  }
 
  if (mutex_init(&client->send_mutex, "client_send") != 0) {
    log_error("Failed to initialize send mutex for client %s", new_client_id);
    remove_client(server_ctx, new_client_id);
    return NULL;
  }
 
  // Register all atomic fields for debug tracking - DISABLED to prevent deadlock
  // register_client_info_atomics() and audio_ring_buffer_register_atomics() call
  // NAMED_REGISTER which tries to acquire WRITE lock on named_registry.entries_lock
  // The debug_sync thread holds READ lock while printing, causing reader-writer deadlock
  // These are debug-only features not critical for functionality
  // log_info("[TCP_DBG] BEFORE_REGISTER_CLIENT_ATOMICS");
  // register_client_info_atomics(client);
  // log_info("[TCP_DBG] AFTER_REGISTER_CLIENT_ATOMICS");
 
  // Audio buffer atomics registration disabled - same deadlock reason
  // log_info("[TCP_DBG] AUDIO_BUFFER_REGISTER_ATOMICS_START: client=%p", (void *)client);
  // if (client->incoming_audio_buffer) {
  //   audio_ring_buffer_register_atomics(client->incoming_audio_buffer, new_client_id);
  //   log_info("[TCP_DBG] AUDIO_BUFFER_REGISTER_ATOMICS_DONE");
  // } else {
  //   log_info("[TCP_DBG] AUDIO_BUFFER_REGISTER_ATOMICS_SKIPPED: incoming_audio_buffer is NULL");
  // }
 
  // Register with audio mixer OUTSIDE lock
  // CRITICAL: Do this BEFORE crypto handshake to avoid deadlock with mixer thread
  log_info("[TCP_DBG] MIXER_ADD_SOURCE_START: g_audio_mixer=%p, client->incoming_audio_buffer=%p",
           (void *)g_audio_mixer, (void *)client->incoming_audio_buffer);
  if (g_audio_mixer && client->incoming_audio_buffer) {
    log_info("[TCP_DBG] MIXER_ADD_SOURCE_CALLING: client_id=%s", new_client_id);
    if (mixer_add_source(g_audio_mixer, new_client_id, client->incoming_audio_buffer) < 0) {
      log_warn("Failed to add client %s to audio mixer", new_client_id);
    } else {
      log_info("[TCP_DBG] MIXER_ADD_SOURCE_SUCCESS: client_id=%s", new_client_id);
#ifdef DEBUG_AUDIO
      log_debug("Added client %s to audio mixer", new_client_id);
#endif
    }
  } else {
    log_info("[TCP_DBG] MIXER_ADD_SOURCE_SKIPPED: conditions not met");
  }
  log_info("[TCP_DBG] MIXER_ADD_SOURCE_DONE");
 
  // Create the client's persistent transport early (before handshake).
  // It starts with NULL crypto; server_crypto_handshake() sets crypto_ctx after success.
  client->transport = acip_tcp_transport_create(new_client_id, socket, NULL);
  if (!client->transport) {
    log_error("Failed to create ACIP transport for client %s", new_client_id);
    if (remove_client(server_ctx, new_client_id) != 0) {
      log_error("Failed to remove client after transport creation failure");
    }
    return NULL;
  }
  log_debug("Created ACIP transport for client %s (crypto pending handshake)", new_client_id);
 
  // Perform crypto handshake before starting threads.
  // This ensures the handshake uses the socket directly without interference from receive thread.
  // The handshake uses client->transport and sets its crypto_ctx on success.
  log_info("[TCP_DBG] CRYPTO_INIT_START: About to call server_crypto_init()");
  if (server_crypto_init() == 0) {
    log_info("[TCP_DBG] CRYPTO_INIT_DONE: server_crypto_init() returned 0");
    // Set timeout for crypto handshake to prevent indefinite blocking
    // This prevents clients from connecting but never completing the handshake
    const uint64_t HANDSHAKE_TIMEOUT_NS = 30ULL * NS_PER_SEC_INT;
    asciichat_error_t timeout_result = set_socket_timeout(socket, HANDSHAKE_TIMEOUT_NS);
    if (timeout_result != ASCIICHAT_OK) {
      log_warn("Failed to set handshake timeout for client %s: %s", new_client_id,
               asciichat_error_string(timeout_result));
      // Continue anyway - timeout is a safety feature, not critical
    }
 
    log_info("[TCP_DBG] SERVER_CRYPTO_HANDSHAKE_START: About to call server_crypto_handshake()");
    int crypto_result = server_crypto_handshake(client);
    log_info("[TCP_DBG] SERVER_CRYPTO_HANDSHAKE_DONE: result=%d", crypto_result);
    if (crypto_result != 0) {
      log_error("Crypto handshake failed for client %s: %s", new_client_id, network_error_string());
      if (remove_client(server_ctx, new_client_id) != 0) {
        log_error("Failed to remove client after crypto handshake failure");
      }
      return NULL;
    }
 
    // Clear socket timeout after handshake completes successfully
    // This allows normal operation without timeouts on data transfer
    log_info("[TCP_DBG] CLEAR_SOCKET_TIMEOUT_START");
    asciichat_error_t clear_timeout_result = set_socket_timeout(socket, 0);
    if (clear_timeout_result != ASCIICHAT_OK) {
      log_warn("Failed to clear handshake timeout for client %s: %s", new_client_id,
               asciichat_error_string(clear_timeout_result));
      // Continue anyway - we can still communicate even with timeout set
    }
 
    log_debug("Crypto handshake completed successfully for client %s", new_client_id);
 
    // After handshake completes, the client immediately sends PACKET_TYPE_CLIENT_CAPABILITIES
    // We must read and process this packet BEFORE starting the receive thread to avoid a race condition
    // where the packet arrives but no thread is listening for it.
    //
    // SPECIAL CASE: If client used --no-encrypt, we already received this packet during the handshake
    // attempt and stored it in pending_packet_*. Use that instead of receiving a new one.
    packet_envelope_t envelope;
    bool used_pending_packet = false;
 
    if (client->pending_packet_payload) {
      // Client used --no-encrypt mode - use the packet we already received
      log_info("Client %s using --no-encrypt mode - processing pending packet type %u", client->client_id,
               client->pending_packet_type);
      envelope.type = client->pending_packet_type;
      envelope.data = client->pending_packet_payload;
      envelope.len = client->pending_packet_length;
      envelope.allocated_buffer = client->pending_packet_payload; // Will be freed below
      envelope.allocated_size = client->pending_packet_length;
      used_pending_packet = true;
 
      // Clear pending packet fields
      client->pending_packet_type = 0;
      client->pending_packet_payload = NULL;
      client->pending_packet_length = 0;
    } else {
      // Normal encrypted mode - receive capabilities packet
      log_debug("Waiting for initial capabilities packet from client %s", client->client_id);
 
      // Protect crypto context access with client state mutex
      mutex_lock(&client->client_state_mutex);
      const crypto_context_t *crypto_ctx = crypto_server_get_context(client->client_id);
 
      // Use per-client crypto state to determine enforcement
      // At this point, handshake is complete, so crypto_initialized=true and handshake is ready
      bool enforce_encryption = !GET_OPTION(no_encrypt) && client->crypto_initialized &&
                                crypto_handshake_is_ready(&client->crypto_handshake_ctx);
 
      packet_recv_result_t result = receive_packet_secure(socket, (void *)crypto_ctx, enforce_encryption, &envelope);
      mutex_unlock(&client->client_state_mutex);
 
      if (result != PACKET_RECV_SUCCESS) {
        log_error("Failed to receive initial capabilities packet from client %s: result=%d", client->client_id, result);
        if (envelope.allocated_buffer) {
          buffer_pool_free(NULL, envelope.allocated_buffer, envelope.allocated_size);
        }
        if (remove_client(server_ctx, client->client_id) != 0) {
          log_error("Failed to remove client after crypto handshake failure");
        }
        return NULL;
      }
    }
 
    if (envelope.type != PACKET_TYPE_CLIENT_CAPABILITIES) {
      log_error("Expected PACKET_TYPE_CLIENT_CAPABILITIES but got packet type %d from client %s", envelope.type,
                client->client_id);
      if (envelope.allocated_buffer) {
        buffer_pool_free(NULL, envelope.allocated_buffer, envelope.allocated_size);
      }
      if (remove_client(server_ctx, client->client_id) != 0) {
        log_error("Failed to remove client after crypto handshake failure");
      }
      return NULL;
    }
 
    // The unencrypted socket receive path retains the wire header in its
    // allocation. Packet handlers always consume payload-only data, matching
    // the encrypted and ACIP-dispatch paths.
    if (envelope.data == envelope.allocated_buffer) {
      const packet_header_t *header = (const packet_header_t *)envelope.allocated_buffer;
      envelope.type = (packet_type_t)NET_TO_HOST_U16(header->type);
      envelope.data = (uint8_t *)envelope.allocated_buffer + sizeof(*header);
      envelope.len = NET_TO_HOST_U32(header->length);
    }
 
    // Process the capabilities packet directly
    log_debug("Processing initial capabilities packet from client %s (from %s)", client->client_id,
              used_pending_packet ? "pending packet" : "network");
    log_info("INITIAL_CAPABILITIES: client_id=%s source=%s type=%u payload_len=%zu allocated_size=%zu data=%p",
             client->client_id, used_pending_packet ? "pending" : "network", envelope.type, envelope.len,
             envelope.allocated_size, envelope.data);
    handle_client_capabilities_packet(client, envelope.data, envelope.len);
 
    // Free the packet data
    if (envelope.allocated_buffer) {
      buffer_pool_free(NULL, envelope.allocated_buffer, envelope.allocated_size);
    }
    log_debug("Successfully received and processed initial capabilities for client %s", client->client_id);
  }
 
  // Start all client threads in the correct order (unified path for TCP and WebRTC)
  // This creates: receive thread -> render threads -> send thread
  // The render threads MUST be created before send thread to avoid the race condition
  // where send thread reads empty frames before render thread generates the first real frame
  char client_id_snapshot[MAX_CLIENT_ID_LEN];
  SAFE_STRNCPY(client_id_snapshot, client->client_id, sizeof(client_id_snapshot) - 1);
  if (start_client_threads(server_ctx, client, true) != 0) {
    log_error("Failed to start threads for TCP client %s", client_id_snapshot);
    // Client is already in hash table - use remove_client for proper cleanup
    remove_client(server_ctx, client_id_snapshot);
    return NULL;
  }
  log_debug("Successfully created render threads for client %s", client_id_snapshot);
 
  // Register client with session_host (for discovery mode support)
  if (server_ctx->session_host) {
    client->session_client_id = session_host_add_client(server_ctx->session_host, socket, client_ip, port);
    if (client->session_client_id == 0) {
      log_warn("Failed to register client %s with session_host", client_id_snapshot);
    } else {
      log_debug("Client %s registered with session_host as %u", client_id_snapshot, client->session_client_id);
    }
  }
 
  // Broadcast server state to ALL clients AFTER the new client is fully set up
  // This notifies all clients (including the new one) about the updated grid
  log_info("[TCP_DBG] BROADCAST_SERVER_STATE_START: About to broadcast to all clients");
  broadcast_server_state_to_all_clients();
  log_info("[TCP_DBG] BROADCAST_SERVER_STATE_DONE");
 
  // Register client atomics for debug sync inspection
  register_client_info_atomics(client);
 
  log_info("[TCP_DBG] ADD_CLIENT_RETURNING: client_id=%s, client=%p", client->client_id, (void *)client);
  return client;
 
error_cleanup:
  // Clean up all partially allocated resources
  // NOTE: This label is reached when allocation or initialization fails BEFORE
  // the client is added to the hash table. Don't call remove_client() here.
  cleanup_client_all_buffers(client);
  rwlock_wrunlock(&g_client_manager_rwlock);
  return NULL;
}

References acip_tcp_transport_create(), client_info::active, packet_envelope_t::allocated_buffer, packet_envelope_t::allocated_size, ASCIICHAT_OK, atomic_load_bool(), atomic_store_bool(), atomic_store_u64(), client_info::audio_queue, audio_ring_buffer_create_for_capture(), audio_ring_buffer_destroy(), broadcast_server_state_to_all_clients(), buffer_pool_free(), client_manager_t::client_count, client_info::client_id, client_info::client_ip, client_info::client_state_mutex, client_manager_t::clients, client_manager_t::clients_by_id, client_info::codec_capabilities_audio, client_info::codec_capabilities_video, client_info::connected_at, client_info::crypto_handshake_ctx, crypto_handshake_is_ready(), client_info::crypto_initialized, crypto_server_get_context(), packet_envelope_t::data, client_info::display_name, errno, ERROR_INTERNAL, ERROR_MEMORY, g_audio_mixer, g_client_manager, g_client_manager_rwlock, generate_client_id(), generate_client_name(), GET_OPTION, handle_client_capabilities_packet(), client_info::incoming_audio_buffer, client_info::incoming_video_buffer, client_info::is_tcp_client, client_info::last_rendered_grid_sources, client_info::last_sent_grid_sources, packet_envelope_t::len, packet_header_t::length, log_debug, LOG_ERRNO_IF_SET, log_error, log_info, log_warn, MAX_CLIENT_ID_LEN, MAX_CLIENTS, MAX_FRAME_BUFFER_SIZE, mixer_add_source(), mutex_init(), mutex_lock, mutex_unlock, NAMED_REGISTER_CLIENT, NET_TO_HOST_U16, NET_TO_HOST_U32, network_error_string(), NS_PER_SEC_INT, client_info::outgoing_video_buffer, packet_queue_create(), packet_queue_destroy(), PACKET_RECV_SUCCESS, PACKET_TYPE_CLIENT_CAPABILITIES, client_info::pending_packet_length, client_info::pending_packet_payload, client_info::pending_packet_type, client_info::port, receive_packet_secure(), remove_client(), rwlock_wrlock, rwlock_wrunlock, SAFE_FREE, SAFE_MALLOC_ALIGNED, safe_snprintf(), SAFE_STRERROR, SAFE_STRNCPY, client_info::send_buffer, client_info::send_buffer_size, client_info::send_mutex, server_crypto_handshake(), server_crypto_init(), client_info::server_ctx, client_info::session_client_id, server_context_t::session_host, session_host_add_client(), SET_ERRNO, set_socket_timeout(), client_info::shutting_down, client_info::socket, socket_send(), server_context_t::tcp_server, tcp_server_add_client(), client_info::transport, client_info::transport_encrypted, packet_header_t::type, packet_envelope_t::type, video_frame_buffer_create(), and video_frame_buffer_destroy().

add_webrtc_client()

client_info_t * add_webrtc_client	(	server_context_t *	server_ctx,
		acip_transport_t *	transport,
		const char *	client_ip,
		bool	start_threads
	)

Register a WebRTC client with the server.

Registers a client that connected via WebRTC data channel instead of TCP socket. This function reuses most of add_client() logic but skips:

• Crypto handshake (already done via ACDS signaling)
• Socket-specific configuration
• TCP thread pool registration

DIFFERENCES FROM add_client():

• Takes an already-created acip_transport_t* instead of socket
• No crypto handshake (WebRTC signaling handled authentication)
• No socket configuration (WebRTC handles buffering)
• Uses generic thread spawning instead of tcp_server thread pool

Parameters

server_ctx	Server context
transport	WebRTC transport (already created and connected)
client_ip	Client IP address for logging (may be empty for P2P)

Returns

Pointer to client_info_t on success, NULL on failure

Note

The transport must be fully initialized and ready to send/receive

Client capabilities are still expected as first packet

Definition at line 1109 of file src/server/client.c.

                                                     {
  if (!server_ctx || !transport || !client_ip) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameters to add_webrtc_client");
    return NULL;
  }
 
  rwlock_wrlock(&g_client_manager_rwlock);
 
  // Find empty slot - this is the authoritative check
  int slot = -1;
  int existing_count = 0;
  for (int i = 0; i < MAX_CLIENTS; i++) {
    if (slot == -1 && g_client_manager.clients[i].client_id[0] == '\0') {
      slot = i; // Take first available slot
    }
    // Count only active clients
    if (g_client_manager.clients[i].client_id[0] != '\0' && atomic_load_bool(&g_client_manager.clients[i].active)) {
      existing_count++;
    }
  }
 
  // Check if we've hit the configured max-clients limit (not the array size)
  if (existing_count >= GET_OPTION(max_clients)) {
    rwlock_wrunlock(&g_client_manager_rwlock);
    SET_ERRNO(ERROR_RESOURCE_EXHAUSTED, "Maximum client limit reached (%d/%d active clients)", existing_count,
              GET_OPTION(max_clients));
    log_error("Maximum client limit reached (%d/%d active clients)", existing_count, GET_OPTION(max_clients));
    return NULL;
  }
 
  if (slot == -1) {
    rwlock_wrunlock(&g_client_manager_rwlock);
    SET_ERRNO(ERROR_RESOURCE_EXHAUSTED, "No available client slots (all %d array slots are in use)", MAX_CLIENTS);
    log_error("No available client slots (all %d array slots are in use)", MAX_CLIENTS);
    return NULL;
  }
 
  // Update client_count to match actual count before adding new client
  g_client_manager.client_count = existing_count;
 
  // Generate unique client ID for WebRTC client (noun only, without counter or port)
  char new_client_id[MAX_CLIENT_ID_LEN];
  if (generate_client_id(new_client_id, sizeof(new_client_id)) != 0) {
    rwlock_wrunlock(&g_client_manager_rwlock);
    log_error("Failed to generate unique client ID for WebRTC client");
    return NULL;
  }
 
  // Initialize client
  client_info_t *client = &g_client_manager.clients[slot];
 
  // Free any existing buffers from previous client in this slot
  if (client->incoming_video_buffer) {
    video_frame_buffer_destroy(client->incoming_video_buffer);
    client->incoming_video_buffer = NULL;
  }
  if (client->outgoing_video_buffer) {
    video_frame_buffer_destroy(client->outgoing_video_buffer);
    client->outgoing_video_buffer = NULL;
  }
  if (client->incoming_audio_buffer) {
    audio_ring_buffer_destroy(client->incoming_audio_buffer);
    client->incoming_audio_buffer = NULL;
  }
  if (client->send_buffer) {
    SAFE_FREE(client->send_buffer);
    client->send_buffer = NULL;
  }
 
  memset(client, 0, sizeof(client_info_t));
 
  // Set up WebRTC-specific fields
  client->socket = INVALID_SOCKET_VALUE; // WebRTC has no traditional socket
  client->is_tcp_client = false;         // WebRTC client - threads managed directly
  client->transport_encrypted = acip_transport_get_type(transport) == ACIP_TRANSPORT_WEBRTC;
  client->transport = transport; // Use provided transport
  SAFE_STRNCPY(client->client_id, new_client_id, sizeof(client->client_id) - 1);
  SAFE_STRNCPY(client->client_ip, client_ip, sizeof(client->client_ip) - 1);
  client->port = 0; // WebRTC doesn't use port numbers
  atomic_store_bool(&client->active, true);
  client->server_ctx = server_ctx; // Store server context for receive thread cleanup
  log_info("Added new WebRTC client %s from %s (transport=%p, slot=%d)", new_client_id, client_ip, transport, slot);
  atomic_store_bool(&client->shutting_down, false);
  atomic_store_u64(&client->last_rendered_grid_sources, 0); // Render thread updates this
  atomic_store_u64(&client->last_sent_grid_sources, 0);     // Send thread updates this
  log_debug("WebRTC client slot assigned: client_id=%s assigned to slot %d", client->client_id, slot);
  client->connected_at = time(NULL);
 
  // Initialize crypto context for this client
  memset(&client->crypto_handshake_ctx, 0, sizeof(client->crypto_handshake_ctx));
  log_debug("[ADD_WEBRTC_CLIENT] Setting crypto_initialized=false (will be set to true after handshake completes)");
  // Do NOT set crypto_initialized = true here! The handshake must complete first.
  // For WebSocket clients: websocket_client_handler will perform the handshake
  // For WebRTC clients: ACDS signaling may have already done crypto (future enhancement)
  client->crypto_initialized = false;
 
  // Initialize pending packet storage (unused for WebRTC, but keep for consistency)
  client->pending_packet_type = 0;
  client->pending_packet_payload = NULL;
  client->pending_packet_length = 0;
 
  // Use noun-based client name for display_name as well
  SAFE_STRNCPY(client->display_name, new_client_id, sizeof(client->display_name) - 1);
 
  // Create individual video buffer for this client using modern double-buffering
  client->incoming_video_buffer = video_frame_buffer_create(new_client_id);
  if (!client->incoming_video_buffer) {
    SET_ERRNO(ERROR_MEMORY, "Failed to create video buffer for WebRTC client %s", new_client_id);
    log_error("Failed to create video buffer for WebRTC client %s", new_client_id);
    goto error_cleanup_webrtc;
  }
 
  // Create individual audio buffer for this client
  client->incoming_audio_buffer = audio_ring_buffer_create_for_capture();
  if (!client->incoming_audio_buffer) {
    SET_ERRNO(ERROR_MEMORY, "Failed to create audio buffer for WebRTC client %s", new_client_id);
    log_error("Failed to create audio buffer for WebRTC client %s", new_client_id);
    goto error_cleanup_webrtc;
  }
 
  // Create packet queues for outgoing data
  client->audio_queue = packet_queue_create_with_pools(500, 1000, false);
  if (!client->audio_queue) {
    LOG_ERRNO_IF_SET("Failed to create audio queue for WebRTC client");
    goto error_cleanup_webrtc;
  }
 
  // Create packet queue for async dispatch: received packets waiting to be processed
  // Capacity of 100 packets allows buffering of ~10-50 frames (depending on fragmentation)
  client->received_packet_queue = packet_queue_create_with_pools(100, 500, false);
  if (!client->received_packet_queue) {
    LOG_ERRNO_IF_SET("Failed to create received packet queue for client");
    goto error_cleanup_webrtc;
  }
 
  // Create outgoing video buffer for ASCII frames (double buffered, no dropping)
  client->outgoing_video_buffer = video_frame_buffer_create(new_client_id);
  if (!client->outgoing_video_buffer) {
    LOG_ERRNO_IF_SET("Failed to create outgoing video buffer for WebRTC client");
    goto error_cleanup_webrtc;
  }
 
  // Pre-allocate send buffer to avoid malloc/free in send thread (prevents deadlocks)
  client->send_buffer_size = MAX_FRAME_BUFFER_SIZE; // 2MB should handle largest frames
  client->send_buffer = SAFE_MALLOC_ALIGNED(client->send_buffer_size, 64, void *);
  if (!client->send_buffer) {
    log_error("Failed to allocate send buffer for WebRTC client %s", new_client_id);
    goto error_cleanup_webrtc;
  }
 
  g_client_manager.client_count = existing_count + 1; // We just added a client
  log_debug("Client count updated: now %d clients (added WebRTC client_id=%s to slot %d)",
            g_client_manager.client_count, new_client_id, slot);
 
  // Add client to uthash table for O(1) lookup
  HASH_ADD_STR(g_client_manager.clients_by_id, client_id, client);
  log_debug("Added WebRTC client %s to uthash table", new_client_id);
 
  // Release the write lock IMMEDIATELY after adding to hash table
  // All subsequent operations (mutex init, mixer registration) don't need global client manager lock
  rwlock_wrunlock(&g_client_manager_rwlock);
 
  // Register this client's audio buffer with the mixer
  if (g_audio_mixer && client->incoming_audio_buffer) {
    if (mixer_add_source(g_audio_mixer, new_client_id, client->incoming_audio_buffer) < 0) {
      log_warn("Failed to add WebRTC client %s to audio mixer", new_client_id);
    } else {
#ifdef DEBUG_AUDIO
      log_debug("Added WebRTC client %s to audio mixer", new_client_id);
#endif
    }
  }
 
  // Initialize mutexes BEFORE creating any threads to prevent race conditions
  if (mutex_init(&client->client_state_mutex, "client_state") != 0) {
    log_error("Failed to initialize client state mutex for WebRTC client %s", new_client_id);
    // Client is already in hash table - use remove_client for proper cleanup
    remove_client(server_ctx, new_client_id);
    return NULL;
  }
 
  initialize_default_render_capabilities(client);
 
  // Initialize send mutex to protect concurrent socket writes
  if (mutex_init(&client->send_mutex, "client_send") != 0) {
    log_error("Failed to initialize send mutex for WebRTC client %s", new_client_id);
    // Client is already in hash table - use remove_client for proper cleanup
    remove_client(server_ctx, client->client_id);
    return NULL;
  }
 
  // Initialize condition variable for dispatch thread wake-up
  if (cond_init(&client->dispatch_queue_cond, "dispatch_queue_cond") != 0) {
    log_error("Failed to initialize dispatch queue condition variable for WebRTC client %s", new_client_id);
    // Client is already in hash table - use remove_client for proper cleanup
    remove_client(server_ctx, client->client_id);
    return NULL;
  }
 
  // For WebRTC clients, the capabilities packet will be received by the receive thread
  // when it starts. Unlike TCP clients where we handle it synchronously in add_client(),
  // WebRTC uses the transport abstraction which handles packet reception automatically.
  log_debug("WebRTC client %s initialized - receive thread will process capabilities", new_client_id);
 
  // Send initial server state to the new client
  if (send_server_state_to_client(client) != 0) {
    log_warn("Failed to send initial server state to WebRTC client %s", new_client_id);
  } else {
#ifdef DEBUG_NETWORK
    log_info("Sent initial server state to WebRTC client %s", new_client_id);
#endif
  }
 
  // Send initial server state via ACIP transport
  rwlock_rdlock(&g_client_manager_rwlock);
  uint32_t connected_count = g_client_manager.client_count;
  rwlock_rdunlock(&g_client_manager_rwlock);
 
  server_state_packet_t state;
  state.connected_client_count = connected_count;
  state.active_client_count = 0; // Will be updated by broadcast thread
  memset(state.reserved, 0, sizeof(state.reserved));
 
  // Convert to network byte order
  server_state_packet_t net_state;
  net_state.connected_client_count = HOST_TO_NET_U32(state.connected_client_count);
  net_state.active_client_count = HOST_TO_NET_U32(state.active_client_count);
  memset(net_state.reserved, 0, sizeof(net_state.reserved));
 
  asciichat_error_t packet_send_result = acip_send_server_state(client->transport, &net_state);
  if (packet_send_result != ASCIICHAT_OK) {
    log_warn("Failed to send initial server state to WebRTC client %s: %s", new_client_id,
             asciichat_error_string(packet_send_result));
  } else {
    log_debug("Sent initial server state to WebRTC client %s: %u connected clients", new_client_id,
              state.connected_client_count);
  }
 
  // Register client with session_host (for discovery mode support)
  // WebRTC clients use INVALID_SOCKET_VALUE since they don't have a TCP socket
  if (server_ctx->session_host) {
    client->session_client_id = session_host_add_client(server_ctx->session_host, INVALID_SOCKET_VALUE, client_ip, 0);
    if (client->session_client_id == 0) {
      log_warn("Failed to register WebRTC client %s with session_host", new_client_id);
    } else {
      log_debug("WebRTC client %s registered with session_host as %u", new_client_id, client->session_client_id);
    }
  }
 
  // Broadcast server state to ALL clients AFTER the new client is fully set up
  // This notifies all clients (including the new one) about the updated grid
  broadcast_server_state_to_all_clients();
 
  // The browser can send CLIENT_CAPABILITIES and STREAM_START immediately when
  // its DataChannel opens. Start media threads only after this client is fully
  // registered and all initial control traffic is complete; the WebRTC
  // transport queues those incoming packets until the receive thread begins.
  // WebSocket callers still defer this until their crypto setup is complete.
  if (start_threads) {
    log_debug("[ADD_WEBRTC_CLIENT] Starting client threads after initialization for %s", new_client_id);
    if (start_client_threads(server_ctx, client, false) != 0) {
      log_error("Failed to start threads for WebRTC client %s", new_client_id);
      return NULL;
    }
  } else {
    log_debug("[ADD_WEBRTC_CLIENT] Deferring client thread startup for %s until crypto setup", new_client_id);
  }
 
  return client;
 
error_cleanup_webrtc:
  // Clean up all partially allocated resources for WebRTC client
  // NOTE: This label is reached when allocation or initialization fails BEFORE
  // the client is added to the hash table. Don't call remove_client() here.
  cleanup_client_all_buffers(client);
  rwlock_wrunlock(&g_client_manager_rwlock);
  return NULL;
}

References acip_send_server_state(), ACIP_TRANSPORT_WEBRTC, client_info::active, server_state_packet_t::active_client_count, ASCIICHAT_OK, atomic_load_bool(), atomic_store_bool(), atomic_store_u64(), client_info::audio_queue, audio_ring_buffer_create_for_capture(), audio_ring_buffer_destroy(), broadcast_server_state_to_all_clients(), client_manager_t::client_count, client_info::client_id, client_info::client_ip, client_info::client_state_mutex, client_manager_t::clients, client_manager_t::clients_by_id, cond_init(), client_info::connected_at, server_state_packet_t::connected_client_count, client_info::crypto_handshake_ctx, client_info::crypto_initialized, client_info::dispatch_queue_cond, client_info::display_name, ERROR_INVALID_PARAM, ERROR_MEMORY, ERROR_RESOURCE_EXHAUSTED, g_audio_mixer, g_client_manager, g_client_manager_rwlock, generate_client_id(), GET_OPTION, HOST_TO_NET_U32, client_info::incoming_audio_buffer, client_info::incoming_video_buffer, INVALID_SOCKET_VALUE, client_info::is_tcp_client, client_info::last_rendered_grid_sources, client_info::last_sent_grid_sources, log_debug, LOG_ERRNO_IF_SET, log_error, log_info, log_warn, MAX_CLIENT_ID_LEN, MAX_CLIENTS, MAX_FRAME_BUFFER_SIZE, mixer_add_source(), mutex_init(), client_info::outgoing_video_buffer, packet_queue_create_with_pools(), client_info::pending_packet_length, client_info::pending_packet_payload, client_info::pending_packet_type, client_info::port, client_info::received_packet_queue, remove_client(), server_state_packet_t::reserved, rwlock_rdlock, rwlock_rdunlock, rwlock_wrlock, rwlock_wrunlock, SAFE_FREE, SAFE_MALLOC_ALIGNED, SAFE_STRNCPY, client_info::send_buffer, client_info::send_buffer_size, client_info::send_mutex, send_server_state_to_client(), client_info::server_ctx, client_info::session_client_id, server_context_t::session_host, session_host_add_client(), SET_ERRNO, client_info::shutting_down, client_info::socket, client_info::transport, client_info::transport_encrypted, video_frame_buffer_create(), and video_frame_buffer_destroy().

broadcast_server_state_to_all_clients()

void broadcast_server_state_to_all_clients

(

void

)

Notify all clients of state changes.

Definition at line 2692 of file src/server/client.c.

                                                 {
  // SNAPSHOT PATTERN: Collect client data while holding lock, then release before network I/O
  typedef struct {
    socket_t socket;
    char client_id[MAX_CLIENT_ID_LEN];
    const crypto_context_t *crypto_ctx;
  } client_snapshot_t;
 
  client_snapshot_t client_snapshots[MAX_CLIENTS];
  int snapshot_count = 0;
  int active_video_count = 0;
 
  uint64_t lock_start_ns = time_get_ns();
  rwlock_rdlock(&g_client_manager_rwlock);
  uint64_t lock_end_ns = time_get_ns();
  uint64_t lock_time_ns = time_elapsed_ns(lock_start_ns, lock_end_ns);
  if (lock_time_ns > 1 * NS_PER_MS_INT) {
    char duration_str[32];
    time_pretty((uint64_t)((double)lock_time_ns), -1, duration_str, sizeof(duration_str));
    log_warn("broadcast_server_state: rwlock_rdlock took %s", duration_str);
  }
 
  // Count active clients and snapshot client data while holding lock
  // Use atomic_load for all atomic fields to prevent data races.
  for (int i = 0; i < MAX_CLIENTS; i++) {
    bool is_active = atomic_load_bool(&g_client_manager.clients[i].active);
    if (is_active && atomic_load_bool(&g_client_manager.clients[i].is_sending_video)) {
      active_video_count++;
    }
    if (is_active && g_client_manager.clients[i].socket != INVALID_SOCKET_VALUE) {
      // Skip clients that haven't completed crypto handshake yet
      // Check both crypto_initialized AND crypto context (defense in depth)
      if (!GET_OPTION(no_encrypt) && !g_client_manager.clients[i].crypto_initialized) {
        log_debug("Skipping server_state broadcast to client %s: crypto handshake not complete",
                  g_client_manager.clients[i].client_id);
        continue;
      }
 
      // Get crypto context (may be NULL if --no-encrypt)
      const crypto_context_t *crypto_ctx =
          crypto_handshake_get_context(&g_client_manager.clients[i].crypto_handshake_ctx);
      if (!GET_OPTION(no_encrypt) && !crypto_ctx) {
        // Skip clients without valid crypto context
        log_debug("Skipping server_state broadcast to client %s: no crypto context",
                  g_client_manager.clients[i].client_id);
        continue;
      }
 
      client_snapshots[snapshot_count].socket = g_client_manager.clients[i].socket;
      SAFE_STRNCPY(client_snapshots[snapshot_count].client_id, g_client_manager.clients[i].client_id,
                   sizeof(client_snapshots[snapshot_count].client_id) - 1);
      client_snapshots[snapshot_count].crypto_ctx = crypto_ctx;
      snapshot_count++;
    }
  }
 
  // Prepare server state packet while still holding lock (fast operation)
  server_state_packet_t state;
  state.connected_client_count = g_client_manager.client_count;
  state.active_client_count = active_video_count;
  memset(state.reserved, 0, sizeof(state.reserved));
 
  // Convert to network byte order
  server_state_packet_t net_state;
  net_state.connected_client_count = HOST_TO_NET_U32(state.connected_client_count);
  net_state.active_client_count = HOST_TO_NET_U32(state.active_client_count);
  memset(net_state.reserved, 0, sizeof(net_state.reserved));
 
  // Release lock BEFORE sending (snapshot pattern)
  // Sending while holding lock blocks all client operations
  uint64_t lock_held_final_ns = time_get_ns();
  uint64_t lock_held_ns = time_elapsed_ns(lock_start_ns, lock_held_final_ns);
  rwlock_rdunlock(&g_client_manager_rwlock);
 
  // Send to all clients AFTER releasing the lock
  // This prevents blocking other threads during network I/O
  for (int i = 0; i < snapshot_count; i++) {
    log_debug_every(5000 * US_PER_MS_INT,
                    "BROADCAST_DEBUG: Sending SERVER_STATE to client %s (socket %d) with crypto_ctx=%p",
                    client_snapshots[i].client_id, client_snapshots[i].socket, (void *)client_snapshots[i].crypto_ctx);
 
    // Protect socket write with per-client send_mutex.
    client_info_t *target = find_client_by_id(client_snapshots[i].client_id);
    if (target) {
      // IMPORTANT: Verify client_id matches expected value - prevents use-after-free
      // if client was removed and replaced with another client in same slot
      if (strcmp(target->client_id, client_snapshots[i].client_id) != 0) {
        log_warn("Client %s ID mismatch during broadcast (found %s), skipping send", client_snapshots[i].client_id,
                 target->client_id);
        continue;
      }
 
      mutex_lock(&target->send_mutex);
 
      // Double-check client_id again after acquiring mutex (stronger protection)
      if (strcmp(target->client_id, client_snapshots[i].client_id) != 0) {
        mutex_unlock(&target->send_mutex);
        log_warn("Client %s was removed during broadcast send (now %s), skipping", client_snapshots[i].client_id,
                 target->client_id);
        continue;
      }
 
      // Send via ACIP transport
      asciichat_error_t result = acip_send_server_state(target->transport, &net_state);
      mutex_unlock(&target->send_mutex);
 
      if (result != ASCIICHAT_OK) {
        log_error("Failed to send server state to client %s: %s", client_snapshots[i].client_id,
                  asciichat_error_string(result));
      } else {
        log_debug_every(5000 * US_PER_MS_INT, "Sent server state to client %s: %u connected, %u active",
                        client_snapshots[i].client_id, state.connected_client_count, state.active_client_count);
      }
    } else {
      log_warn("Client %s removed before broadcast send could complete", client_snapshots[i].client_id);
    }
  }
 
  if (lock_held_ns > 1 * NS_PER_MS_INT) {
    char duration_str[32];
    time_pretty((uint64_t)((double)lock_held_ns), -1, duration_str, sizeof(duration_str));
    log_warn("broadcast_server_state: rwlock held for %s (includes network I/O)", duration_str);
  }
}

References acip_send_server_state(), client_info::active, server_state_packet_t::active_client_count, ASCIICHAT_OK, atomic_load_bool(), client_manager_t::client_count, client_info::client_id, client_manager_t::clients, server_state_packet_t::connected_client_count, client_info::crypto_handshake_ctx, crypto_handshake_get_context(), client_info::crypto_initialized, find_client_by_id(), g_client_manager, g_client_manager_rwlock, GET_OPTION, HOST_TO_NET_U32, INVALID_SOCKET_VALUE, client_info::is_sending_video, log_debug, log_debug_every, log_error, log_warn, MAX_CLIENT_ID_LEN, MAX_CLIENTS, mutex_lock, mutex_unlock, NS_PER_MS_INT, server_state_packet_t::reserved, rwlock_rdlock, rwlock_rdunlock, SAFE_STRNCPY, client_info::send_mutex, client_info::socket, time_elapsed_ns(), time_get_ns(), time_pretty(), client_info::transport, and US_PER_MS_INT.

Referenced by add_client(), add_webrtc_client(), and remove_client().

cleanup_client_media_buffers()

void cleanup_client_media_buffers

(

client_info_t *

client

)

Definition at line 2873 of file src/server/client.c.

                                                         {
  if (!client) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Client is NULL");
    return;
  }
 
  if (client->digital_rain) {
    digital_rain_destroy(client->digital_rain);
    client->digital_rain = NULL;
  }
 
  if (client->incoming_video_buffer) {
    video_frame_buffer_destroy(client->incoming_video_buffer);
    client->incoming_video_buffer = NULL;
  }
 
  // Clean up outgoing video buffer (for ASCII frames)
  if (client->outgoing_video_buffer) {
    video_frame_buffer_destroy(client->outgoing_video_buffer);
    client->outgoing_video_buffer = NULL;
  }
 
  // Clean up pre-allocated send buffer
  if (client->send_buffer) {
    SAFE_FREE(client->send_buffer);
    client->send_buffer = NULL;
    client->send_buffer_size = 0;
  }
 
  if (client->incoming_audio_buffer) {
    audio_ring_buffer_destroy(client->incoming_audio_buffer);
    client->incoming_audio_buffer = NULL;
  }
 
  // Clean up Opus decoder
  if (client->opus_decoder) {
    opus_codec_destroy((opus_codec_t *)client->opus_decoder);
    client->opus_decoder = NULL;
  }
 
  // Clean up H.265 decoder (persistent per-client decoder)
  if (client->h265_decoder_ctx) {
    avcodec_free_context((AVCodecContext **)&client->h265_decoder_ctx);
    client->h265_decoder_ctx = NULL;
  }
  if (client->h265_decode_frame) {
    av_frame_free((AVFrame **)&client->h265_decode_frame);
    client->h265_decode_frame = NULL;
  }
  if (client->h265_rgb_frame) {
    av_frame_free((AVFrame **)&client->h265_rgb_frame);
    client->h265_rgb_frame = NULL;
  }
  if (client->h265_sws_ctx) {
    sws_freeContext(client->h265_sws_ctx);
    client->h265_sws_ctx = NULL;
  }
}

References audio_ring_buffer_destroy(), client_info::digital_rain, digital_rain_destroy(), ERROR_INVALID_PARAM, client_info::h265_decode_frame, client_info::h265_decoder_ctx, client_info::h265_rgb_frame, client_info::h265_sws_ctx, client_info::incoming_audio_buffer, client_info::incoming_video_buffer, opus_codec_destroy(), client_info::opus_decoder, client_info::outgoing_video_buffer, SAFE_FREE, client_info::send_buffer, client_info::send_buffer_size, SET_ERRNO, and video_frame_buffer_destroy().

cleanup_client_packet_queues()

void cleanup_client_packet_queues

(

client_info_t *

client

)

Definition at line 2932 of file src/server/client.c.

                                                         {
  if (!client)
    return;
 
  if (client->audio_queue) {
    packet_queue_destroy(client->audio_queue);
    client->audio_queue = NULL;
  }
 
  // Async dispatch: clean up received packet queue
  if (client->received_packet_queue) {
    packet_queue_destroy(client->received_packet_queue);
    client->received_packet_queue = NULL;
  }
 
  // Video now uses double buffer, cleaned up in cleanup_client_media_buffers
}

References client_info::audio_queue, packet_queue_destroy(), and client_info::received_packet_queue.

client_dispatch_thread()

void * client_dispatch_thread

(

void *

arg

)

Async dispatch thread for WebRTC clients.

Async dispatch thread - processes queued received packets.

Decouples receive from dispatch to prevent backpressure on the network socket. Received packets are queued by the receive thread, and processed asynchronously here. This allows the receive thread to keep accepting packets while dispatch processes them.

Definition at line 1706 of file src/server/client.c.

                                        {
  client_info_t *client = (client_info_t *)arg;
 
  if (!client) {
    log_error("Invalid client info in dispatch thread (NULL pointer)");
    return NULL;
  }
 
  const char *client_id = client->client_id;
  log_info("DISPATCH_THREAD: Started for client %s", client_id);
 
  uint64_t dispatch_loop_count = 0;
  queued_packet_t *pending_packets[128] = {0};
  size_t pending_count = 0;
  size_t pending_index = 0;
 
  while (!atomic_load_bool(&g_should_exit) && atomic_load_bool(&client->dispatch_thread_running)) {
    dispatch_loop_count++;
    // Try to dequeue next packet (non-blocking)
    // Use try_dequeue to avoid blocking - allows checking exit flag frequently
    uint64_t dequeue_start = time_get_ns();
    mutex_lock(&client->client_state_mutex);
    if (pending_index == pending_count) {
      pending_count = 0;
      pending_index = 0;
      queued_packet_t *packet;
      while (pending_count < 128 && (packet = packet_queue_try_dequeue(client->received_packet_queue))) {
        pending_packets[pending_count++] = packet;
      }
      // Audio packets interleave with raw video. Coalesce snapshots across audio
      // without crossing control or encoded-frame boundaries.
      bool newer_image = false;
      for (size_t i = pending_count; i > 0; i--) {
        packet_type_t type = NET_TO_HOST_U16(pending_packets[i - 1]->header.type);
        if (type == PACKET_TYPE_IMAGE_FRAME) {
          if (newer_image) {
            packet_queue_free_packet(pending_packets[i - 1]);
            pending_packets[i - 1] = NULL;
          }
          newer_image = true;
        } else if (type != PACKET_TYPE_AUDIO_BATCH && type != PACKET_TYPE_AUDIO_OPUS_BATCH) {
          newer_image = false;
        }
      }
    }
    while (pending_index < pending_count && !pending_packets[pending_index]) pending_index++;
    queued_packet_t *queued_pkt = pending_index < pending_count ? pending_packets[pending_index++] : NULL;
    uint64_t dequeue_end = time_get_ns();
 
    if (!queued_pkt) {
      // Queue was empty, wait for signal from receive thread or timeout
      // Use 100ms timeout to handle cases where signal is missed due to race conditions
      log_dev_every(5 * US_PER_MS_INT, "🔄 DISPATCH_LOOP[%llu]: Queue empty after %.1fμs, waiting for signal",
                    (unsigned long long)dispatch_loop_count, (dequeue_end - dequeue_start) / 1000.0);
      cond_timedwait(&client->dispatch_queue_cond, &client->client_state_mutex, 100 * 1000 * 1000); // 100ms timeout
      mutex_unlock(&client->client_state_mutex);
      continue;
    }
    mutex_unlock(&client->client_state_mutex);
 
    log_dev("DISPATCH_THREAD[%s]: dequeued %zu-byte packet in %.1fμs", client_id, queued_pkt->data_len,
            (dequeue_end - dequeue_start) / 1000.0);
 
    // Process the dequeued packet
    // The queued packet contains the complete ACIP packet (header + payload) from websocket_recv()
    const packet_header_t *header = (const packet_header_t *)queued_pkt->data;
    size_t total_len = queued_pkt->data_len;
    uint8_t *payload = (uint8_t *)header + sizeof(packet_header_t);
    size_t payload_len = 0;
 
    log_dev("DISPATCH_THREAD[%s]: processing %zu-byte packet", client_id, total_len);
 
    if (total_len < sizeof(packet_header_t)) {
      log_error("🔴 DISPATCH_THREAD[%s]: Packet too small (%zu < %zu), DROPPING", client_id, total_len,
                sizeof(packet_header_t));
      packet_queue_free_packet(queued_pkt);
      continue;
    }
 
    {
      packet_type_t packet_type = (packet_type_t)NET_TO_HOST_U16(header->type);
      payload_len = NET_TO_HOST_U32(header->length);
 
    log_dev("DISPATCH_THREAD[%s]: type=%d payload_len=%u total_len=%zu", client_id, packet_type, payload_len,
              total_len);
 
      // Handle PACKET_TYPE_ENCRYPTED from WebSocket clients that encrypt at application layer
      // This mirrors the decryption logic in acip_server_receive_and_dispatch()
      if (packet_type == PACKET_TYPE_ENCRYPTED && client->transport && client->transport->crypto_ctx) {
        log_dev("DISPATCH_THREAD[%s]: decrypting packet", client_id);
 
        uint8_t *ciphertext = payload;
        size_t ciphertext_len = payload_len;
 
        // Decrypt to get inner plaintext packet (header + payload)
        size_t plaintext_size = ciphertext_len + 1024;
        uint8_t *plaintext = SAFE_MALLOC(plaintext_size, uint8_t *);
        if (!plaintext) {
          log_error("🔴 DISPATCH_THREAD[%s]: Failed to allocate plaintext buffer for decryption", client_id);
          packet_queue_free_packet(queued_pkt);
          continue;
        }
 
        size_t plaintext_len;
        crypto_result_t crypto_result = crypto_decrypt(client->transport->crypto_ctx, ciphertext, ciphertext_len,
                                                       plaintext, plaintext_size, &plaintext_len);
 
        if (crypto_result != CRYPTO_OK) {
          log_error("🔴 DISPATCH_THREAD[%s]: Failed to decrypt packet: %s", client_id,
                    crypto_result_to_string(crypto_result));
          SAFE_FREE(plaintext);
          packet_queue_free_packet(queued_pkt);
          continue;
        }
 
        if (plaintext_len < sizeof(packet_header_t)) {
          log_error("🔴 DISPATCH_THREAD[%s]: Decrypted packet too small: %zu < %zu", client_id, plaintext_len,
                    sizeof(packet_header_t));
          SAFE_FREE(plaintext);
          packet_queue_free_packet(queued_pkt);
          continue;
        }
 
        // Parse the inner (decrypted) header
        const packet_header_t *inner_header = (const packet_header_t *)plaintext;
        packet_type = (packet_type_t)NET_TO_HOST_U16(inner_header->type);
        payload_len = NET_TO_HOST_U32(inner_header->length);
        payload = plaintext + sizeof(packet_header_t);
 
        log_dev("DISPATCH_THREAD[%s]: decrypted type=%d payload_len=%u", client_id, packet_type, payload_len);
 
        // Dispatch the decrypted packet
        if (client->transport) {
          log_dev("DISPATCH_THREAD[%s]: dispatching type=%d payload_len=%u", client_id, packet_type, payload_len);
          asciichat_error_t dispatch_result = acip_handle_server_packet(client->transport, packet_type, payload,
                                                                        payload_len, client, &g_acip_server_callbacks);
 
          if (dispatch_result != ASCIICHAT_OK) {
            log_error("🔴 DISPATCH_THREAD[%s]: Handler failed for decrypted packet type=%d: %s", client_id, packet_type,
                      asciichat_error_string(dispatch_result));
          } else {
            log_dev("DISPATCH_THREAD[%s]: dispatched decrypted type=%d", client_id, packet_type);
          }
        } else {
          log_error("🔴 DISPATCH_THREAD[%s]: Cannot dispatch decrypted packet - transport is NULL", client_id);
        }
 
        // Free the decrypted buffer
        SAFE_FREE(plaintext);
      } else {
        // Not encrypted or no crypto context - dispatch as-is
        if (client->transport) {
          log_dev("DISPATCH_THREAD[%s]: dispatching type=%d payload_len=%zu", client_id, packet_type, payload_len);
          asciichat_error_t dispatch_result = acip_handle_server_packet(client->transport, packet_type, payload,
                                                                        payload_len, client, &g_acip_server_callbacks);
 
          if (dispatch_result != ASCIICHAT_OK) {
            log_error("🔴 DISPATCH_THREAD[%s]: Handler failed for packet type=%d: %s", client_id, packet_type,
                      asciichat_error_string(dispatch_result));
          } else {
            log_dev("DISPATCH_THREAD[%s]: dispatched type=%d", client_id, packet_type);
          }
        } else {
          log_error("🔴 DISPATCH_THREAD[%s]: Cannot dispatch packet - transport is NULL", client_id);
        }
      }
    }
 
    // Free the queued packet
    packet_queue_free_packet(queued_pkt);
  }
 
  while (pending_index < pending_count) {
    packet_queue_free_packet(pending_packets[pending_index++]);
  }
  log_info("DISPATCH_THREAD: Exiting for client %s", client_id);
  return NULL;
}

References acip_handle_server_packet(), ASCIICHAT_OK, atomic_load_bool(), client_info::client_id, client_info::client_state_mutex, cond_timedwait, acip_transport::crypto_ctx, crypto_decrypt(), CRYPTO_OK, crypto_result_to_string(), queued_packet_t::data, queued_packet_t::data_len, client_info::dispatch_queue_cond, client_info::dispatch_thread_running, g_should_exit, packet_header_t::length, log_dev, log_dev_every, log_error, log_info, mutex_lock, mutex_unlock, NET_TO_HOST_U16, NET_TO_HOST_U32, packet_queue_free_packet(), packet_queue_try_dequeue(), PACKET_TYPE_AUDIO_BATCH, PACKET_TYPE_AUDIO_OPUS_BATCH, PACKET_TYPE_ENCRYPTED, PACKET_TYPE_IMAGE_FRAME, client_info::received_packet_queue, SAFE_FREE, SAFE_MALLOC, time_get_ns(), client_info::transport, packet_header_t::type, and US_PER_MS_INT.

client_receive_thread()

void * client_receive_thread

(

void *

arg

)

Definition at line 1885 of file src/server/client.c.

                                       {
  // Log thread startup
  log_debug("RECV_THREAD: Thread function entered, arg=%p", arg);
 
  client_info_t *client = (client_info_t *)arg;
 
  // Validate client pointer immediately before any access.
  // This prevents crashes if remove_client() has zeroed the client struct
  // while the thread was still starting at RtlUserThreadStart.
  if (!client) {
    log_error("Invalid client info in receive thread (NULL pointer)");
    return NULL;
  }
 
  // Save this thread's ID so remove_client() can detect self-joins
  client->receive_thread_id = asciichat_thread_self();
 
  log_debug("RECV_THREAD_DEBUG: Thread started, client=%p, client_id=%s, is_tcp=%d", (void *)client, client->client_id,
            client->is_tcp_client);
 
  if (atomic_load_bool(&client->protocol_disconnect_requested)) {
    log_debug("Receive thread for client %s exiting before start (protocol disconnect requested)", client->client_id);
    return NULL;
  }
 
  // Check if client_id is empty (client struct has been zeroed by remove_client)
  // This must be checked BEFORE accessing any client fields
  if (client->client_id[0] == 0) {
    log_debug("Receive thread: client_id is empty, client struct may have been zeroed, exiting");
    return NULL;
  }
 
  // Additional validation: check socket is valid
  // For TCP clients, validate socket. WebRTC clients use DataChannel (no socket)
  if (client->is_tcp_client && client->socket == INVALID_SOCKET_VALUE) {
    log_error("Invalid client socket in receive thread");
    return NULL;
  }
 
  // NOTE: Atomic registration disabled
  // register_client_info_atomics() and audio_ring_buffer_register_atomics() attempt
  // to acquire WRITE lock on named_registry.entries_lock, which conflicts with debug_sync
  // thread holding READ lock during printing, causing deadlock.
  // These are debug-only features, not critical for functionality.
  // The atomics work fine without being registered in the debug system.
 
  // Enable thread cancellation for clean shutdown
  // Thread cancellation not available in platform abstraction
  // Threads should exit when g_should_exit is set
 
  log_debug("Started receive thread for client %s (%s)", client->client_id, client->display_name);
 
  // Main receive loop - processes packets from transport
  // For TCP clients: receives from socket
  // For WebRTC clients: receives from transport ringbuffer (via ACDS signaling)
 
  log_info("RECV_THREAD_LOOP_START: client_id=%s, is_tcp=%d, transport=%p, active=%d", client->client_id,
           client->is_tcp_client, (void *)client->transport, atomic_load_bool(&client->active));
 
  while (!atomic_load_bool(&g_should_exit) && atomic_load_bool(&client->active)) {
    // For TCP clients, check socket validity
    // For WebRTC clients, continue even if no socket (transport handles everything)
    if (client->is_tcp_client && client->socket == INVALID_SOCKET_VALUE) {
      log_debug("TCP client %s has invalid socket, exiting receive thread", client->client_id);
      break;
    }
 
    // Check client_id is still valid before accessing transport.
    // This prevents accessing freed memory if remove_client() has zeroed the client struct.
    if (client->client_id[0] == 0) {
      log_debug("Client client_id reset, exiting receive thread");
      break;
    }
 
    // Receive packet (without dispatching) - decouple from dispatch for async processing
    // For WebRTC clients with async dispatch, we queue packets instead of processing immediately
    // This prevents backpressure on the network socket
 
    if (client->is_tcp_client) {
      // TCP clients: use original synchronous dispatch
      asciichat_error_t acip_result =
          acip_server_receive_and_dispatch(client->transport, client, &g_acip_server_callbacks);
 
      // Check if shutdown was requested during the network call
      if (atomic_load_bool(&g_should_exit)) {
        log_debug("RECV_EXIT: Server shutdown requested, breaking loop");
        break;
      }
 
      // Handle receive errors
      if (acip_result != ASCIICHAT_OK) {
        asciichat_error_context_t err_ctx;
        if (HAS_ERRNO(&err_ctx)) {
          log_error("🔴 ACIP error for client %s: code=%u msg=%s", client->client_id, err_ctx.code,
                    err_ctx.context_message);
          if (err_ctx.code == ERROR_NETWORK) {
            log_debug("Client %s disconnected (network error): %s", client->client_id, err_ctx.context_message);
            break;
          } else if (err_ctx.code == ERROR_CRYPTO) {
            log_error_client(
                client, "SECURITY VIOLATION: Unencrypted packet when encryption required - terminating connection");
            atomic_store_bool(&g_should_exit, true);
            break;
          }
        }
        log_warn("ACIP error for TCP client %s: %s (disconnecting)", client->client_id,
                 asciichat_error_string(acip_result));
        break;
      }
    } else {
      // WebRTC/WebSocket clients: async dispatch - receive packet and queue for async processing
      void *packet_data = NULL;
      void *allocated_buffer = NULL;
      size_t packet_len = 0;
 
      // Snapshot transport pointer to prevent use-after-free
      // Transport can be destroyed by WebSocket callback while we're receiving
      acip_transport_t *transport_snapshot = NULL;
      mutex_lock(&client->send_mutex);
      if (!client->transport || atomic_load_bool(&client->shutting_down)) {
        mutex_unlock(&client->send_mutex);
        log_debug("RECV_THREAD[%s]: Transport destroyed or client shutting down, exiting receive loop",
                  client->client_id);
        break;
      }
      transport_snapshot = client->transport;
      mutex_unlock(&client->send_mutex);
 
      log_debug("🔍 RECV_THREAD[%s]: About to call transport->recv() (transport=%p)", client->client_id,
                (void *)transport_snapshot);
 
      asciichat_error_t recv_result =
          transport_snapshot->methods->recv(transport_snapshot, &packet_data, &packet_len, &allocated_buffer);
 
      // Check error BEFORE logging - logging system may clear thread-local errno
      if (recv_result != ASCIICHAT_OK) {
        asciichat_error_context_t err_ctx;
        if (HAS_ERRNO(&err_ctx)) {
          // Check for reassembly timeout (fragments arriving slowly)
          // This is NOT a connection failure - safe to retry
          if ((err_ctx.code == ERROR_NETWORK) && err_ctx.context_message &&
              strstr(err_ctx.context_message, "reassembly timeout")) {
            // Fragments are arriving slowly - this is normal, retry without disconnecting
            log_dev_every(100000, "Client %s: fragment reassembly timeout, retrying in 10ms", client->client_id);
            APP_CALLBACK_VOID(platform_pump_events);
            platform_sleep_ms(10); // Sleep 10ms to allow fragments to arrive
            continue;              // Retry without disconnecting
          }
 
          if (err_ctx.code == ERROR_NETWORK) {
            log_debug("Client %s disconnected (network error): %s", client->client_id, err_ctx.context_message);
            break;
          }
        }
        log_warn("Receive failed for WebRTC client %s: %s (disconnecting)", client->client_id,
                 asciichat_error_string(recv_result));
        break;
      }
 
      log_dev("RECV_THREAD[%s]: recv result=%d packet_len=%zu", client->client_id, recv_result, packet_len);
 
      // Validate received packet before queueing
      if (packet_len < sizeof(packet_header_t)) {
        log_warn("RECV_THREAD[%s]: Received packet too small (%zu < %zu), dropping", client->client_id, packet_len,
                 sizeof(packet_header_t));
        if (allocated_buffer) {
          buffer_pool_free(NULL, allocated_buffer, packet_len);
        }
        continue;
      }
 
      // Queue the received packet for async dispatch
      // This prevents the receive thread from blocking on dispatch
      log_dev("RECV_THREAD[%s]: queueing %zu-byte packet", client->client_id, packet_len);
 
      // Extract packet type from the header to preserve it when queueing
      const packet_header_t *pkt_header = (const packet_header_t *)allocated_buffer;
      packet_type_t pkt_type = (packet_type_t)NET_TO_HOST_U16(pkt_header->type);
      uint32_t payload_len = NET_TO_HOST_U32(pkt_header->length);
 
      log_dev("RECV_THREAD[%s]: type=%d payload_len=%u total_len=%zu", client->client_id, pkt_type, payload_len,
              packet_len);
 
      // Build a complete packet to queue (header + payload)
      // The entire buffer (allocated_buffer) contains the full packet
      // The queue copies the packet; release the transport buffer afterward.
      uint32_t client_id_hash = fnv1a_hash_string(client->client_id);
      // Pair queue publication and notification with the dispatch wait lock.
      mutex_lock(&client->client_state_mutex);
      int enqueue_result = packet_queue_enqueue(client->received_packet_queue, pkt_type, allocated_buffer, packet_len,
                                                client_id_hash, true);
      if (enqueue_result >= 0)
        cond_signal(&client->dispatch_queue_cond);
      mutex_unlock(&client->client_state_mutex);
      buffer_pool_free(NULL, allocated_buffer, packet_len);
 
      if (enqueue_result < 0) {
        log_error("🔴 RECV_THREAD[%s]: Failed to queue received packet (queue full?) - DROPPING FRAME",
                  client->client_id);
      } else {
        log_dev("RECV_THREAD[%s]: queued type=%d len=%zu", client->client_id, pkt_type, packet_len);
      }
    }
  }
 
  // Mark client as inactive and stop all threads
  // Must stop render threads when client disconnects.
  // OPTIMIZED: Use atomic operations for thread control flags (lock-free)
  char client_id_snapshot[MAX_CLIENT_ID_LEN] = {0};
  SAFE_STRNCPY(client_id_snapshot, client->client_id, sizeof(client_id_snapshot) - 1);
  bool is_tcp_client = client->is_tcp_client;
  server_context_t *server_ctx = (server_context_t *)client->server_ctx;
  log_debug("Setting active=false in receive_thread_fn (client_id=%s, exiting receive loop)", client_id_snapshot);
  atomic_store_bool(&client->active, false);
  atomic_store_bool(&client->send_thread_running, false);
  atomic_store_bool(&client->video_render_thread_running, false);
  atomic_store_bool(&client->audio_render_thread_running, false);
 
  // Call remove_client() to trigger cleanup
  // Safe to call from receive thread now: remove_client() detects self-join via thread IDs
  // and skips the receive thread join when called from the receive thread itself
  log_debug("Receive thread for client %s calling remove_client() for cleanup", client_id_snapshot);
  if (server_ctx && !is_tcp_client) {
    if (remove_client(server_ctx, client_id_snapshot) != 0) {
      log_warn("Failed to remove client %s from receive thread cleanup", client_id_snapshot);
    }
  } else if (!server_ctx) {
    log_error("Receive thread for client %s: server_ctx is NULL, cannot call remove_client()", client_id_snapshot);
  }
 
  log_debug("Receive thread for client %s terminated", client_id_snapshot);
 
  // Clean up thread-local error context before exit
  asciichat_errno_destroy();
 
  return NULL;
}

References acip_server_receive_and_dispatch(), client_info::active, APP_CALLBACK_VOID, asciichat_errno_destroy(), ASCIICHAT_OK, asciichat_thread_self, atomic_load_bool(), atomic_store_bool(), client_info::audio_render_thread_running, buffer_pool_free(), client_info::client_id, client_info::client_state_mutex, asciichat_error_context_t::code, cond_signal(), asciichat_error_context_t::context_message, client_info::dispatch_queue_cond, client_info::display_name, ERROR_CRYPTO, ERROR_NETWORK, g_should_exit, HAS_ERRNO, INVALID_SOCKET_VALUE, client_info::is_tcp_client, packet_header_t::length, log_debug, log_dev, log_dev_every, log_error, log_error_client, log_info, log_warn, MAX_CLIENT_ID_LEN, acip_transport::methods, mutex_lock, mutex_unlock, NET_TO_HOST_U16, NET_TO_HOST_U32, packet_queue_enqueue(), platform_sleep_ms(), client_info::protocol_disconnect_requested, client_info::receive_thread_id, client_info::received_packet_queue, acip_transport_methods_t::recv, remove_client(), SAFE_STRNCPY, client_info::send_mutex, client_info::send_thread_running, client_info::server_ctx, client_info::shutting_down, client_info::socket, client_info::transport, packet_header_t::type, and client_info::video_render_thread_running.

client_send_thread_func()

void * client_send_thread_func

(

void *

arg

)

Client packet send thread.

Definition at line 2196 of file src/server/client.c.

                                         {
  client_info_t *client = (client_info_t *)arg;
 
  // Validate client pointer immediately before any access.
  // This prevents crashes if remove_client() has zeroed the client struct
  // while the thread was still starting at RtlUserThreadStart.
  if (!client) {
    log_error("Invalid client info in send thread (NULL pointer)");
    return NULL;
  }
 
  client->send_thread_id = asciichat_thread_self();
 
  // Check if client_id is empty (client struct has been zeroed by remove_client)
  // This must be checked BEFORE accessing any client fields
  if (client->client_id[0] == 0) {
    log_debug_every(100 * US_PER_MS_INT,
                    "Send thread: client_id is empty, client struct may have been zeroed, exiting");
    return NULL;
  }
 
  // Additional validation: check socket OR transport is valid
  // For TCP clients: socket is valid
  // For WebRTC clients: socket is INVALID_SOCKET_VALUE but transport is valid
  mutex_lock(&client->send_mutex);
  bool has_socket = (client->socket != INVALID_SOCKET_VALUE);
  bool has_transport = (client->transport != NULL);
  mutex_unlock(&client->send_mutex);
 
  log_info("SEND_THREAD_VALIDATION: client_id=%s socket_valid=%d transport_valid=%d transport_ptr=%p",
           client->client_id, has_socket, has_transport, (void *)client->transport);
 
  if (!has_socket && !has_transport) {
    log_error("Invalid client connection in send thread (no socket or transport)");
    return NULL;
  }
 
  log_info("Started send thread for client %s (%s)", client->client_id, client->display_name);
 
  // Mark thread as running
  atomic_store_bool(&client->send_thread_running, true);
 
  log_info("SEND_THREAD_LOOP_START: client_id=%s active=%d shutting_down=%d running=%d", client->client_id,
           atomic_load_bool(&client->active), atomic_load_bool(&client->shutting_down),
           atomic_load_bool(&client->send_thread_running));
 
  // Track timing for video frame sends
  uint64_t last_video_send_time = 0;
  const uint64_t video_send_interval_us = 16666; // 60fps = ~16.67ms
  const int target_video_fps = client->has_terminal_caps && client->terminal_caps.desired_fps > 0
                                   ? client->terminal_caps.desired_fps
                                   : VIDEO_RENDER_FPS;
 
  // High-frequency audio loop - separate from video frame loop
  // to ensure audio packets are sent immediately, not rate-limited by video
// A video deadline must still be reachable while audio is active. Sending a
// deep audio batch on the same ordered WebRTC channel lets one client hold
// video (and the next audio update) behind several synchronous sends.
  int loop_iteration_count = 0;
  bool transport_failed = false;
  uint64_t video_rate_window_start_ns = time_get_ns();
  unsigned long video_rate_window_start_count = atomic_load_u64(&client->frames_sent_count);
  while (!atomic_load_bool(&g_should_exit) && !atomic_load_bool(&client->shutting_down) &&
         atomic_load_bool(&client->active) && atomic_load_bool(&client->send_thread_running)) {
    loop_iteration_count++;
    bool sent_something = false;
    uint64_t loop_start_ns = time_get_ns();
    log_dev_every(5000 * US_PER_MS_INT, "[SEND_LOOP_%d] START: client=%s", loop_iteration_count, client->client_id);
 
    // PRIORITY: Drain all queued audio packets before video
    // Audio must not be rate-limited by video frame sending (16.67ms)
    queued_packet_t *audio_packets[MAX_AUDIO_BATCH];
    int audio_packet_count = 0;
 
    if (client->audio_queue) {
      // Try to dequeue multiple audio packets
      for (int i = 0; i < MAX_AUDIO_BATCH; i++) {
        audio_packets[i] = packet_queue_try_dequeue(client->audio_queue);
        if (audio_packets[i]) {
          audio_packet_count++;
        } else {
          break; // No more packets available
        }
      }
      if (audio_packet_count > 0) {
        log_dev_every(4500 * US_PER_MS_INT, "SEND_AUDIO: client=%s dequeued=%d packets", client->client_id,
                      audio_packet_count);
      }
    } else {
      log_warn("Send thread: audio_queue is NULL for client %s", client->client_id);
    }
 
    // Send batched audio if we have packets
    if (audio_packet_count > 0) {
      // Protect crypto field access with mutex
      mutex_lock(&client->client_state_mutex);
      bool crypto_ready = client->transport_encrypted || GET_OPTION(no_encrypt) ||
                          (client->crypto_initialized && crypto_handshake_is_ready(&client->crypto_handshake_ctx));
      mutex_unlock(&client->client_state_mutex);
 
      // Audio packets cannot be sent until crypto handshake completes (protocol requirement)
      if (!crypto_ready) {
        log_dev_every(1000 * US_PER_MS_INT, "SKIP_AUDIO: client=%s handshake not ready, dropping %d audio packets",
                      client->client_id, audio_packet_count);
        // Drop audio packets until handshake completes
        for (int i = 0; i < audio_packet_count; i++) {
          packet_queue_free_packet(audio_packets[i]);
        }
        audio_packet_count = 0;
      }
 
      asciichat_error_t result = ASCIICHAT_OK;
 
      // Only send audio if packets survived crypto check (count > 0 after potential drop)
      if (audio_packet_count > 0) {
        uint8_t combined_audio_payload[AUDIO_OPUS_BATCH_HEADER_SIZE + MAX_AUDIO_BATCH *
                                                                     (sizeof(uint16_t) + AUDIO_OPUS_FRAME_MAX_BYTES)];
        const void *audio_payload = audio_packets[0]->data;
        size_t audio_payload_length = audio_packets[0]->data_len;
        if (audio_packet_count > 1) {
          result = combine_audio_opus_packets(audio_packets, audio_packet_count, combined_audio_payload,
                                              sizeof(combined_audio_payload), &audio_payload_length);
          if (result == ASCIICHAT_OK) {
            audio_payload = combined_audio_payload;
          }
        }
 
        mutex_lock(&client->send_mutex);
        acip_transport_t *transport = client->transport;
        bool stopping = atomic_load_bool(&client->shutting_down) || !transport;
        if (stopping) {
          mutex_unlock(&client->send_mutex);
          result = ERROR_NETWORK;
        } else if (result != ASCIICHAT_OK) {
          mutex_unlock(&client->send_mutex);
        } else {
          uint32_t client_id_hash = fnv1a_hash_string(client->client_id);
          result = packet_send_via_transport(transport, PACKET_TYPE_AUDIO_OPUS_BATCH, audio_payload,
                                             audio_payload_length, client_id_hash);
          mutex_unlock(&client->send_mutex);
        }
      } // End of if (audio_packet_count > 0)
 
      // Free all audio packets
      for (int i = 0; i < audio_packet_count; i++) {
        packet_queue_free_packet(audio_packets[i]);
      }
 
      if (result != ASCIICHAT_OK) {
        if (!atomic_load_bool(&g_should_exit)) {
          log_error("Failed to send audio to client %s: %s", client->client_id, asciichat_error_string(result));
        }
        // Network errors corrupt the TCP stream - must disconnect immediately
        if (result == ERROR_NETWORK) {
          transport_failed = true;
          log_error("CLOSING_CLIENT_ON_NETWORK_ERROR: client_id=%s due to audio send failure (corrupted stream)",
                    client->client_id);
          break; // Exit send loop to trigger cleanup
        }
        log_warn("SKIP_AUDIO_ERROR: client_id=%s result=%d (continuing to send video)", client->client_id, result);
        // Continue sending video for non-network errors (audio is optional for browser clients)
      }
 
      sent_something = true;
      uint64_t audio_done_ns = time_get_ns();
      char audio_elapsed_str[32];
      time_pretty(audio_done_ns - loop_start_ns, -1, audio_elapsed_str, sizeof(audio_elapsed_str));
      log_dev_every(5000 * US_PER_MS_INT, "[SEND_LOOP_%d] AUDIO_SENT: took %s", loop_iteration_count,
                     audio_elapsed_str);
 
    } else {
      // No audio packets - brief sleep to avoid busy-looping, then check for other tasks
      log_dev_every(5000 * US_PER_MS_INT, "[SEND_LOOP_%d] NO_AUDIO: sleeping 1ms", loop_iteration_count);
      platform_sleep_us(1 * US_PER_MS_INT); // 1ms - enough for audio render thread to queue more packets
 
      // Check if session rekeying should be triggered
      mutex_lock(&client->client_state_mutex);
      bool should_rekey = !GET_OPTION(no_encrypt) && client->crypto_initialized &&
                          crypto_handshake_is_ready(&client->crypto_handshake_ctx) &&
                          crypto_handshake_should_rekey(&client->crypto_handshake_ctx);
      mutex_unlock(&client->client_state_mutex);
 
      if (should_rekey) {
        log_debug("Rekey threshold reached for client %s, initiating session rekey", client->client_id);
        mutex_lock(&client->client_state_mutex);
        // Get transport reference briefly to avoid deadlock on transport buffers
        mutex_lock(&client->send_mutex);
        if (atomic_load_bool(&client->shutting_down) || !client->transport) {
          mutex_unlock(&client->send_mutex);
          mutex_unlock(&client->client_state_mutex);
          log_warn("BREAK_REKEY: client_id=%s shutting_down=%d transport=%p", client->client_id,
                   atomic_load_bool(&client->shutting_down), (void *)client->transport);
          break; // Client is shutting down, exit thread
        }
        acip_transport_t *rekey_transport = client->transport;
        mutex_unlock(&client->send_mutex);
 
        // Network I/O happens OUTSIDE the send_mutex (client_state_mutex still held for crypto state)
        asciichat_error_t result = crypto_handshake_rekey_request(&client->crypto_handshake_ctx, rekey_transport);
        mutex_unlock(&client->client_state_mutex);
 
        if (result != ASCIICHAT_OK) {
          log_error("Failed to send REKEY_REQUEST to client %s: %d", client->client_id, result);
        } else {
          log_debug("Sent REKEY_REQUEST to client %s", client->client_id);
          // Notify client that session rekeying has been initiated (old keys still active)
          log_info_client(client, "Session rekey initiated - rotating encryption keys");
        }
      }
    }
 
    // Leave the newest frame available until the send deadline. Consuming it
    // before pacing is ready discards frames when render and send clocks drift.
    uint64_t pacing_time_us = time_ns_to_us(time_get_ns());
    if (pacing_time_us - last_video_send_time < video_send_interval_us) {
      platform_sleep_us(1 * US_PER_MS_INT);
      continue;
    }
    uint64_t video_check_ns = time_get_ns();
    if (!client->outgoing_video_buffer) {
      // Buffer has been destroyed (client is shutting down).
      // Exit cleanly instead of looping forever trying to access freed memory.
      log_warn("⚠️  Send thread exiting: outgoing_video_buffer is NULL for client %s (client shutting down?)",
               client->client_id);
      break;
    }
 
    // Check the frame deadline before consuming the latest-frame flag.
    uint64_t current_time_ns = time_get_ns();
    uint64_t current_time_us = time_ns_to_us(current_time_ns);
    uint64_t time_since_last_send_us = current_time_us - last_video_send_time;
    log_dev_every(4500 * US_PER_MS_INT,
                  "Send thread timing check: time_since_last=%llu us, interval=%llu us, should_send=%d",
                  (unsigned long long)time_since_last_send_us, (unsigned long long)video_send_interval_us,
                  (time_since_last_send_us >= video_send_interval_us));
 
    if (current_time_us - last_video_send_time >= video_send_interval_us) {
      video_frame_buffer_t *outgoing_buffer = client->outgoing_video_buffer;
      const video_frame_t *frame = NULL;
      char *frame_data_snapshot = NULL;
      size_t frame_size = 0;
      uint32_t width = 0;
      uint32_t height = 0;
 
      // The frame pointer is borrowed only until the next commit. Keep the
      // swap mutex held through the copy so a slow network send cannot read a
      // buffer after the render thread has reused it.
      mutex_lock(&outgoing_buffer->swap_mutex);
      if (atomic_load_bool(&outgoing_buffer->new_frame_available)) {
        frame = outgoing_buffer->front_buffer;
        if (frame && frame->data && frame->size > 0 &&
            frame->size <= outgoing_buffer->allocated_buffer_size) {
          frame_size = frame->size;
          width = frame->width;
          height = frame->height;
          frame_data_snapshot = SAFE_MALLOC(frame_size, char *);
          if (frame_data_snapshot) {
            memcpy(frame_data_snapshot, frame->data, frame_size);
            atomic_store_bool(&outgoing_buffer->new_frame_available, false);
          }
        }
      }
      mutex_unlock(&outgoing_buffer->swap_mutex);
 
      uint64_t frame_get_ns = time_get_ns();
      char frame_get_elapsed_str[32];
      time_pretty(frame_get_ns - video_check_ns, -1, frame_get_elapsed_str, sizeof(frame_get_elapsed_str));
      log_dev_every(5000 * US_PER_MS_INT, "[SEND_LOOP_%d] VIDEO_SNAPSHOT: took %s, frame=%p, bytes=%zu",
                    loop_iteration_count, frame_get_elapsed_str, (const void *)frame, frame_size);
 
      if (!frame_data_snapshot) {
        // The render thread may not have committed a frame yet, or the latest
        // frame could not be copied. Keep the sender alive and try again.
        log_dev_every(LOG_RATE_SLOW, "Send thread: no snapshot ready for client %s", client->client_id);
        platform_sleep_us(1 * US_PER_MS_INT);
        continue;
      }
 
      log_dev_every(5000 * US_PER_MS_INT, "SEND_TIME_READY: client_id=%s time_since=%llu interval=%llu",
                     client->client_id, (unsigned long long)time_since_last_send_us,
                     (unsigned long long)video_send_interval_us);
      uint64_t frame_start_ns = time_get_ns();
 
      // GRID LAYOUT CHANGE: Check if render thread has buffered a frame with different source count
      // If so, send CLEAR_CONSOLE before sending the new frame
      int rendered_sources = (int)atomic_load_u64(&client->last_rendered_grid_sources);
      int sent_sources = (int)atomic_load_u64(&client->last_sent_grid_sources);
 
      if (rendered_sources != sent_sources && rendered_sources > 0) {
        // Grid layout changed! Send CLEAR_CONSOLE before next frame using ACIP transport
        // Keep the send lock until the complete ACIP packet is written. TCP is
        // a byte stream, so a PONG or state update from another thread must not
        // interleave with this packet.
        mutex_lock(&client->send_mutex);
        if (atomic_load_bool(&client->shutting_down) || !client->transport) {
          mutex_unlock(&client->send_mutex);
          SAFE_FREE(frame_data_snapshot);
          log_warn("BREAK_CLEAR_CONSOLE: client_id=%s shutting_down=%d transport=%p", client->client_id,
                   atomic_load_bool(&client->shutting_down), (void *)client->transport);
          break; // Client is shutting down, exit thread
        }
        acip_transport_t *clear_transport = client->transport;
        acip_send_clear_console(clear_transport);
        mutex_unlock(&client->send_mutex);
        log_debug_every(LOG_RATE_FAST, "Client %s: Sent CLEAR_CONSOLE (grid changed %d → %d sources)",
                        client->client_id, sent_sources, rendered_sources);
        atomic_store_u64(&client->last_sent_grid_sources, rendered_sources);
        sent_something = true;
      }
 
      log_info_every(5000 * US_PER_MS_INT,
                     "🎯 FRAME_VALIDATION: client_id=%s frame=%p, frame->data=%p, frame->size=%zu", client->client_id,
                     (const void *)frame, (void *)frame_data_snapshot, frame_size);
 
      log_info_every(5000 * US_PER_MS_INT, "✅ FRAME_DATA_OK: client_id=%s data=%p", client->client_id,
                     (void *)frame_data_snapshot);
      log_info_every(5000 * US_PER_MS_INT, "✅ FRAME_SIZE_OK: client_id=%s size=%zu", client->client_id, frame_size);
 
      const char *frame_data = frame_data_snapshot;
      uint64_t step1_ns = time_get_ns();
      uint64_t step2_ns = time_get_ns();
      uint64_t step3_ns = time_get_ns();
      uint64_t step4_ns = time_get_ns();
 
      // Check if crypto handshake is complete before sending (prevents sending to unauthenticated clients)
      mutex_lock(&client->client_state_mutex);
      bool crypto_ready = client->transport_encrypted || GET_OPTION(no_encrypt) ||
                          (client->crypto_initialized && crypto_handshake_is_ready(&client->crypto_handshake_ctx));
      mutex_unlock(&client->client_state_mutex);
 
      log_info_every(5000 * US_PER_MS_INT,
                     "🔐 CRYPTO_CHECK: client_id=%s crypto_ready=%d crypto_initialized=%d no_encrypt=%d",
                     client->client_id, crypto_ready, client->crypto_initialized, GET_OPTION(no_encrypt));
 
      if (!crypto_ready) {
        log_info("⚠️  SKIP_SEND_CRYPTO: client_id=%s crypto_initialized=%d no_encrypt=%d", client->client_id,
                 client->crypto_initialized, GET_OPTION(no_encrypt));
        SAFE_FREE(frame_data_snapshot);
        continue; // Skip this frame, will try again on next loop iteration
      }
      log_info_every(5000 * US_PER_MS_INT, "✅ CRYPTO_READY: client_id=%s about to send frame", client->client_id);
 
      // ACIP transport handles header building, CRC32, and encryption. Keep
      // send_mutex for the complete write so packets from the receive thread
      // cannot splice into this TCP packet.
      log_dev_every(4500 * US_PER_MS_INT,
                    "Send thread: About to send frame to client %s (width=%u, height=%u, size=%zu, data=%p)",
                    client->client_id, width, height, frame_size, (void *)frame_data);
 
      // Log first 32 bytes of frame data to verify we can access it
      if (frame_data && frame_size > 0) {
        log_info_every(5000 * US_PER_MS_INT,
                       "FRAME_DATA_HEX: client=%s first_bytes=[%02x %02x %02x %02x %02x %02x %02x %02x %02x %02x %02x "
                       "%02x %02x %02x %02x %02x]",
                       client->client_id, ((uint8_t *)frame_data)[0], ((uint8_t *)frame_data)[1],
                       ((uint8_t *)frame_data)[2], ((uint8_t *)frame_data)[3], ((uint8_t *)frame_data)[4],
                       ((uint8_t *)frame_data)[5], ((uint8_t *)frame_data)[6], ((uint8_t *)frame_data)[7],
                       ((uint8_t *)frame_data)[8], ((uint8_t *)frame_data)[9], ((uint8_t *)frame_data)[10],
                       ((uint8_t *)frame_data)[11], ((uint8_t *)frame_data)[12], ((uint8_t *)frame_data)[13],
                       ((uint8_t *)frame_data)[14], ((uint8_t *)frame_data)[15]);
      }
 
      mutex_lock(&client->send_mutex);
      if (atomic_load_bool(&client->shutting_down) || !client->transport) {
        mutex_unlock(&client->send_mutex);
        SAFE_FREE(frame_data_snapshot);
        log_warn("BREAK_FRAME_SEND: client_id=%s shutting_down=%d transport=%p loop_iter=%d", client->client_id,
                 atomic_load_bool(&client->shutting_down), (void *)client->transport, loop_iteration_count);
        break; // Client is shutting down, exit thread
      }
      acip_transport_t *frame_transport = client->transport;
      log_dev_every(4500 * US_PER_MS_INT, "SEND_ASCII_FRAME: client_id=%s tcp=%d size=%zu width=%u height=%u",
                    client->client_id, client->is_tcp_client, frame_size, width, height);
      uint64_t send_start_ns = time_get_ns();
      log_dev_every(4500 * US_PER_MS_INT, "[SEND_LOOP_%d] FRAME_SEND_START: client_id=%s size=%zu",
                    loop_iteration_count, client->client_id, frame_size);
      asciichat_error_t send_result =
          acip_send_ascii_frame(frame_transport, frame_data, frame_size, width, height, client->client_id);
      mutex_unlock(&client->send_mutex);
      SAFE_FREE(frame_data_snapshot);
      uint64_t send_end_ns = time_get_ns();
      char send_elapsed_str[32];
      time_pretty(send_end_ns - send_start_ns, -1, send_elapsed_str, sizeof(send_elapsed_str));
      log_dev_every(4500 * US_PER_MS_INT, "[SEND_LOOP_%d] FRAME_SEND_END: client_id=%s took %s result=%d",
                    loop_iteration_count, client->client_id, send_elapsed_str, send_result);
      uint64_t step5_ns = time_get_ns();
 
      if (send_result != ASCIICHAT_OK) {
        transport_failed = send_result == ERROR_NETWORK;
        if (!atomic_load_bool(&g_should_exit)) {
          SET_ERRNO(ERROR_NETWORK, "Failed to send video frame to client %s: %s", client->client_id,
                    asciichat_error_string(send_result));
        }
        log_error("SEND_FRAME_FAILED: client_id=%s result=%d message=%s loop_iter=%d", client->client_id, send_result,
                  asciichat_error_string(send_result), loop_iteration_count);
        log_warn("BREAK_SEND_ERROR: client_id=%s (frame send failed)", client->client_id);
        break;
      }
 
      log_dev_every(4500 * US_PER_MS_INT, "SEND_FRAME_SUCCESS: client_id=%s size=%zu", client->client_id, frame_size);
 
      // Increment frame counter for metrics.
      unsigned long frame_count = atomic_fetch_add_u64(&client->frames_sent_count, 1) + 1;
      log_dev_every(4500 * US_PER_MS_INT, "FRAME_SENT: client_id=%s frame_num=%lu size=%zu", client->client_id,
                    frame_count, frame_size);
      uint64_t rate_now_ns = time_get_ns();
      uint64_t rate_elapsed_ns = rate_now_ns - video_rate_window_start_ns;
      if (rate_elapsed_ns >= NS_PER_SEC_INT) {
        unsigned long rate_frames = frame_count - video_rate_window_start_count;
        double delivery_fps = (double)rate_frames * NS_PER_SEC_INT / (double)rate_elapsed_ns;
        if (delivery_fps < (double)target_video_fps * 0.9) {
          log_warn("Server ASCII delivery lagging client=%s fps=%.1f target=%d frames=%lu elapsed=%.2fs",
                   client->client_id, delivery_fps, target_video_fps, rate_frames,
                   (double)rate_elapsed_ns / (double)NS_PER_SEC_INT);
        } else {
          log_info("Server ASCII delivery client=%s fps=%.1f target=%d frames=%lu elapsed=%.2fs", client->client_id,
                   delivery_fps, target_video_fps, rate_frames,
                   (double)rate_elapsed_ns / (double)NS_PER_SEC_INT);
        }
        video_rate_window_start_ns = rate_now_ns;
        video_rate_window_start_count = frame_count;
      }
 
      sent_something = true;
      last_video_send_time = last_video_send_time == 0
                                 ? current_time_us
                                 : current_time_us - (current_time_us - last_video_send_time) % video_send_interval_us;
 
      uint64_t frame_end_ns = time_get_ns();
      uint64_t frame_time_us = time_ns_to_us(time_elapsed_ns(frame_start_ns, frame_end_ns));
      if (frame_time_us > 15 * US_PER_MS_INT) { // Log if sending a frame takes > 15ms (encryption adds ~5-6ms)
        uint64_t step1_us = time_ns_to_us(time_elapsed_ns(frame_start_ns, step1_ns));
        uint64_t step2_us = time_ns_to_us(time_elapsed_ns(step1_ns, step2_ns));
        uint64_t step3_us = time_ns_to_us(time_elapsed_ns(step2_ns, step3_ns));
        uint64_t step4_us = time_ns_to_us(time_elapsed_ns(step3_ns, step4_ns));
        uint64_t step5_us = time_ns_to_us(time_elapsed_ns(step4_ns, step5_ns));
        log_warn_every(
            LOG_RATE_DEFAULT,
            "SEND_THREAD: Frame send took %.2fms for client %s | Snapshot: %.2fms | Memcpy: %.2fms | CRC32: %.2fms | "
            "Header: %.2fms | send_packet_secure: %.2fms",
            frame_time_us / 1000.0, client->client_id, step1_us / 1000.0, step2_us / 1000.0, step3_us / 1000.0,
            step4_us / 1000.0, step5_us / 1000.0);
      }
    }
 
    // If we didn't send anything, sleep briefly to prevent busy waiting
    if (!sent_something) {
      log_dev_every(5000 * US_PER_MS_INT, "[SEND_LOOP_%d] IDLE_SLEEP: nothing sent", loop_iteration_count);
      platform_sleep_us(1 * US_PER_MS_INT); // 1ms sleep
    }
    uint64_t loop_end_ns = time_get_ns();
    uint64_t loop_elapsed_ns = loop_end_ns - loop_start_ns;
    if (loop_elapsed_ns > 10 * NS_PER_MS) {
      char loop_elapsed_str[32];
      time_pretty(loop_elapsed_ns, -1, loop_elapsed_str, sizeof(loop_elapsed_str));
      log_warn_every(LOG_RATE_DEFAULT, "SEND_LOOP: took %s (client=%s)", loop_elapsed_str, client->client_id);
    }
  }
 
  // Log why the send thread exited
  log_warn("Send thread exit conditions - client_id=%s g_should_exit=%d shutting_down=%d active=%d "
           "send_thread_running=%d",
           client->client_id, atomic_load_bool(&g_should_exit), atomic_load_bool(&client->shutting_down),
           atomic_load_bool(&client->active), atomic_load_bool(&client->send_thread_running));
 
  // Mark thread as stopped
  atomic_store_bool(&client->send_thread_running, false);
 
  // The receive loop can remain alive while the media send path has already
  // discovered a dead WebRTC DataChannel. Retire the whole client here so its
  // render workers cannot keep producing media for a peer that is gone.
  if (transport_failed && !client->is_tcp_client && !atomic_load_bool(&g_should_exit) &&
      !atomic_load_bool(&client->shutting_down)) {
    char client_id_snapshot[MAX_CLIENT_ID_LEN] = {0};
    SAFE_STRNCPY(client_id_snapshot, client->client_id, sizeof(client_id_snapshot) - 1);
    server_context_t *server_ctx = (server_context_t *)client->server_ctx;
    if (server_ctx && client_id_snapshot[0] != '\0') {
      log_info("Removing WebRTC client %s after terminal send failure", client_id_snapshot);
      (void)remove_client(server_ctx, client_id_snapshot);
    }
  }
 
  log_debug("Send thread for client %s terminated", client->client_id);
 
  // Clean up thread-local error context before exit
  asciichat_errno_destroy();
 
  return NULL;
}

References acip_send_ascii_frame(), acip_send_clear_console(), client_info::active, video_frame_buffer_t::allocated_buffer_size, asciichat_errno_destroy(), ASCIICHAT_OK, asciichat_thread_self, atomic_fetch_add_u64(), atomic_load_bool(), atomic_load_u64(), atomic_store_bool(), atomic_store_u64(), AUDIO_OPUS_BATCH_HEADER_SIZE, AUDIO_OPUS_FRAME_MAX_BYTES, client_info::audio_queue, client_info::client_id, client_info::client_state_mutex, client_info::crypto_handshake_ctx, crypto_handshake_is_ready(), crypto_handshake_rekey_request(), crypto_handshake_should_rekey(), client_info::crypto_initialized, queued_packet_t::data, queued_packet_t::data_len, terminal_capabilities_t::desired_fps, client_info::display_name, ERROR_NETWORK, frame, client_info::frames_sent_count, video_frame_buffer_t::front_buffer, g_should_exit, GET_OPTION, client_info::has_terminal_caps, INVALID_SOCKET_VALUE, client_info::is_tcp_client, client_info::last_rendered_grid_sources, client_info::last_sent_grid_sources, log_debug, log_debug_every, log_dev_every, log_error, log_info, log_info_client, log_info_every, LOG_RATE_DEFAULT, LOG_RATE_FAST, LOG_RATE_SLOW, log_warn, log_warn_every, MAX_AUDIO_BATCH, MAX_CLIENT_ID_LEN, mutex_lock, mutex_unlock, video_frame_buffer_t::new_frame_available, NS_PER_MS, NS_PER_SEC_INT, client_info::outgoing_video_buffer, packet_queue_free_packet(), packet_queue_try_dequeue(), packet_send_via_transport(), PACKET_TYPE_AUDIO_OPUS_BATCH, platform_sleep_us(), remove_client(), SAFE_FREE, SAFE_MALLOC, SAFE_STRNCPY, client_info::send_mutex, client_info::send_thread_id, client_info::send_thread_running, client_info::server_ctx, SET_ERRNO, client_info::shutting_down, video_frame_t::size, client_info::socket, video_frame_buffer_t::swap_mutex, client_info::terminal_caps, time_elapsed_ns(), time_get_ns(), time_pretty(), client_info::transport, client_info::transport_encrypted, US_PER_MS_INT, and VIDEO_RENDER_FPS.

find_client_by_id()

client_info_t * find_client_by_id

(

const char *

client_id

)

Fast O(1) client lookup by ID using hash table.

This is the primary method for locating clients throughout the server. It uses a hash table for constant-time lookups regardless of client count, making it suitable for high-performance operations like rendering and stats.

PERFORMANCE CHARACTERISTICS:

• Time Complexity: O(1) average case, O(n) worst case (hash collision)
• Space Complexity: O(1)
• Thread Safety: Hash table is internally thread-safe for lookups

USAGE PATTERNS:

• Called by render threads to find target clients for frame generation
• Used by protocol handlers to locate clients for packet processing
• Stats collection for per-client performance monitoring

Parameters

client_id

Unique identifier for the client (0 is invalid)

Returns

Pointer to client_info_t if found, NULL if not found or invalid ID

Note

Does not require external locking - hash table provides thread safety

Returns direct pointer to client struct - caller should use snapshot pattern

Definition at line 429 of file src/server/client.c.

                                                        {
  if (!client_id || client_id[0] == '\0') {
    SET_ERRNO(ERROR_INVALID_PARAM, "Invalid client ID");
    return NULL;
  }
 
  // Protect uthash lookup with read lock to prevent concurrent access issues
  rwlock_rdlock(&g_client_manager_rwlock);
 
  client_info_t *result = NULL;
  HASH_FIND_STR(g_client_manager.clients_by_id, client_id, result);
 
  rwlock_rdunlock(&g_client_manager_rwlock);
 
  if (!result) {
    log_warn("Client not found for ID %s", client_id);
  }
 
  return result;
}

References client_manager_t::clients_by_id, ERROR_INVALID_PARAM, g_client_manager, g_client_manager_rwlock, log_warn, rwlock_rdlock, rwlock_rdunlock, and SET_ERRNO.

Referenced by broadcast_server_state_to_all_clients(), crypto_server_cleanup_client(), crypto_server_decrypt_packet(), crypto_server_encrypt_packet(), crypto_server_get_context(), crypto_server_is_ready(), and start_webrtc_client_threads().

find_client_by_socket()

client_info_t * find_client_by_socket

(

socket_t

socket

)

Find client by socket descriptor using linear search.

This function provides socket-based client lookup, primarily used during connection establishment before client IDs are assigned. Less efficient than find_client_by_id() but necessary for socket-based operations.

PERFORMANCE CHARACTERISTICS:

• Time Complexity: O(n) where n = number of active clients
• Space Complexity: O(1)
• Thread Safety: Internally acquires read lock on g_client_manager_rwlock

USAGE PATTERNS:

• Connection establishment during add_client() processing
• Socket error handling and cleanup operations
• Debugging and diagnostic functions

Parameters

socket

Platform-abstracted socket descriptor to search for

Returns

Pointer to client_info_t if found, NULL if not found

Note

Only searches active clients (avoids returning stale entries)

Caller should use snapshot pattern when accessing returned client data

Definition at line 473 of file src/server/client.c.

                                                      {
  rwlock_rdlock(&g_client_manager_rwlock);
 
  for (int i = 0; i < MAX_CLIENTS; i++) {
    if (g_client_manager.clients[i].socket == socket && atomic_load_bool(&g_client_manager.clients[i].active)) {
      client_info_t *client = &g_client_manager.clients[i];
      rwlock_rdunlock(&g_client_manager_rwlock);
      return client;
    }
  }
 
  rwlock_rdunlock(&g_client_manager_rwlock);
  return NULL;
}

References client_info::active, atomic_load_bool(), client_manager_t::clients, g_client_manager, g_client_manager_rwlock, MAX_CLIENTS, rwlock_rdlock, rwlock_rdunlock, and client_info::socket.

process_decrypted_packet()

void process_decrypted_packet	(	client_info_t *	client,
		packet_type_t	type,
		void *	data,
		size_t	len
	)

Process a decrypted packet from a client

Parameters

client	Client info structure
type	Packet type
data	Packet data
len	Packet length

Definition at line 3934 of file src/server/client.c.

                                                                                                 {
  if (type == 5000) { // CLIENT_CAPABILITIES
    log_debug("CLIENT: client_id=%s, data=%p, len=%zu", client->client_id, data, len);
  }
 
  // Rate limiting: Check and record packet-specific rate limits
  if (g_rate_limiter) {
    if (!check_and_record_packet_rate_limit(g_rate_limiter, client->client_ip, client->socket, type)) {
      // Rate limit exceeded - error response already sent by utility function
      return;
    }
  }
 
  // O(1) dispatch via hash table lookup
  int idx = client_dispatch_hash_lookup(g_client_dispatch_hash, type);
  if (type == 5000 || type == 3001) {
    log_error("DISPATCH_LOOKUP: type=%d, idx=%d (len=%zu)", type, idx, len);
  }
  if (idx < 0) {
    disconnect_client_for_bad_data(client, "Unknown packet type: %d (len=%zu)", type, len);
    return;
  }
 
  if (type == 5000 || type == 3001) {
    log_error("DISPATCH_HANDLER: type=%d, calling handler[%d]...", type, idx);
  }
  g_client_dispatch_handlers[idx](client, data, len);
  if (type == 5000 || type == 3001) {
    log_error("DISPATCH_HANDLER: type=%d, handler returned", type);
  }
}

References check_and_record_packet_rate_limit(), client_info::client_id, client_info::client_ip, disconnect_client_for_bad_data(), g_rate_limiter, log_debug, log_error, and client_info::socket.

process_encrypted_packet()

int process_encrypted_packet	(	client_info_t *	client,
		packet_type_t *	type,
		void **	data,
		size_t *	len,
		uint32_t *	sender_id
	)

Process an encrypted packet from a client

Parameters

client	Client info structure
type	Pointer to packet type (will be updated with decrypted type)
data	Pointer to packet data (will be updated with decrypted data)
len	Pointer to packet length (will be updated with decrypted length)
sender_id	Pointer to sender ID (will be updated with decrypted sender ID)

Returns

0 on success, -1 on error

Definition at line 2974 of file src/server/client.c.

                                                  {
  if (!crypto_server_is_ready(client->client_id)) {
    log_error("Received encrypted packet but crypto not ready for client %s", client->client_id);
    buffer_pool_free(NULL, *data, *len);
    *data = NULL;
    return -1;
  }
 
  // Store original allocation size before it gets modified
  size_t original_alloc_size = *len;
  void *decrypted_data = buffer_pool_alloc(NULL, original_alloc_size);
  size_t decrypted_len;
  int decrypt_result = crypto_server_decrypt_packet(client->client_id, (const uint8_t *)*data, *len,
                                                    (uint8_t *)decrypted_data, original_alloc_size, &decrypted_len);
 
  if (decrypt_result != 0) {
    SET_ERRNO(ERROR_CRYPTO, "Failed to process encrypted packet from client %s (result=%d)", client->client_id,
              decrypt_result);
    buffer_pool_free(NULL, *data, original_alloc_size);
    buffer_pool_free(NULL, decrypted_data, original_alloc_size);
    *data = NULL;
    return -1;
  }
 
  // Replace encrypted data with decrypted data
  // Use original allocation size for freeing the encrypted buffer
  buffer_pool_free(NULL, *data, original_alloc_size);
 
  *data = decrypted_data;
  *len = decrypted_len;
 
  // Now process the decrypted packet by parsing its header
  if (*len < sizeof(packet_header_t)) {
    SET_ERRNO(ERROR_CRYPTO, "Decrypted packet too small for header from client %s", client->client_id);
    buffer_pool_free(NULL, *data, *len);
    *data = NULL;
    return -1;
  }
 
  packet_header_t *header = (packet_header_t *)*data;
  *type = (packet_type_t)NET_TO_HOST_U16(header->type);
  *sender_id = NET_TO_HOST_U32(header->client_id);
 
  // Adjust data pointer to skip header
  *data = (uint8_t *)*data + sizeof(packet_header_t);
  *len -= sizeof(packet_header_t);
 
  return 0;
}

References buffer_pool_alloc(), buffer_pool_free(), client_info::client_id, packet_header_t::client_id, crypto_server_decrypt_packet(), crypto_server_is_ready(), ERROR_CRYPTO, log_error, NET_TO_HOST_U16, NET_TO_HOST_U32, SET_ERRNO, and packet_header_t::type.

remove_client()

int remove_client	(	server_context_t *	server_ctx,
		const char *	client_id
	)

Definition at line 1389 of file src/server/client.c.

                                                                       {
  if (!server_ctx) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Cannot remove client %s: NULL server_ctx", client_id);
    return -1;
  }
 
  // Phase 1: Mark client inactive and prepare for cleanup while holding write lock
  client_info_t *target_client = NULL;
  char display_name_copy[MAX_DISPLAY_NAME_LEN];
  socket_t client_socket = INVALID_SOCKET_VALUE; // Save socket for thread cleanup
 
  log_debug("SOCKET_DEBUG: Attempting to remove client %s", client_id);
  rwlock_wrlock(&g_client_manager_rwlock);
 
  for (int i = 0; i < MAX_CLIENTS; i++) {
    client_info_t *client = &g_client_manager.clients[i];
    if (strcmp(client->client_id, client_id) == 0 && client->client_id[0] != '\0') {
      // Check if already being removed by another thread
      // This prevents double-free and use-after-free crashes during concurrent cleanup
      if (atomic_load_bool(&client->shutting_down)) {
        rwlock_wrunlock(&g_client_manager_rwlock);
        log_debug("Client %s already being removed by another thread, skipping", client_id);
        return 0; // Return success - removal is in progress
      }
      // Mark as shutting down and inactive immediately to stop new operations
      log_debug("Setting active=false in remove_client (client_id=%s, socket=%d)", client_id, client->socket);
      log_info("Removing client %s (socket=%d) - marking inactive and clearing video flags", client_id, client->socket);
      atomic_store_bool(&client->shutting_down, true);
      atomic_store_bool(&client->active, false);
      atomic_store_bool(&client->is_sending_video, false);
      atomic_store_bool(&client->is_sending_audio, false);
      target_client = client;
 
      // Store display name before clearing
      SAFE_STRNCPY(display_name_copy, client->display_name, MAX_DISPLAY_NAME_LEN - 1);
 
      // Save socket for tcp_server_stop_client_threads(). Keep the descriptor
      // open until thread cleanup completes so the OS cannot reuse its fd for
      // a new client while this client's cleanup is still in progress.
      mutex_lock(&client->client_state_mutex);
      client_socket = client->socket; // Save socket for thread cleanup
      mutex_unlock(&client->client_state_mutex);
 
      // Wake blocked receive/send operations without closing the descriptor.
      // Closing here would allow a concurrently accepted client to reuse the
      // same descriptor before the old client's worker threads finish.
      if (client_socket != INVALID_SOCKET_VALUE) {
        (void)socket_shutdown(client_socket, SHUT_RDWR);
      }
 
      // Shutdown packet queues to unblock send thread
      if (client->audio_queue) {
        packet_queue_stop(client->audio_queue);
      }
      // Video now uses double buffer, no queue to shutdown
 
      break;
    }
  }
 
  // If client not found, unlock and return
  if (!target_client) {
    rwlock_wrunlock(&g_client_manager_rwlock);
    log_warn("Cannot remove client %s: not found", client_id);
    return -1;
  }
 
  // Unregister client from session_host (for discovery mode support)
  // NOTE: Client may not be registered if crypto handshake failed before session_host registration
  if (server_ctx->session_host && target_client->session_client_id != 0) {
    asciichat_error_t session_result =
        session_host_remove_client(server_ctx->session_host, target_client->session_client_id);
    if (session_result != ASCIICHAT_OK) {
      // ERROR_NOT_FOUND (91) is expected if client failed crypto before being registered with session_host
      if (session_result == ERROR_NOT_FOUND) {
        log_debug("Client %s not found in session_host (likely failed crypto before registration)", client_id);
      } else {
        log_warn("Failed to unregister client %s from session_host: %s", client_id,
                 asciichat_error_string(session_result));
      }
    } else {
      log_debug("Client %s unregistered from session_host", client_id);
    }
  }
 
  // Release write lock before joining threads.
  // This prevents deadlock with render threads that need read locks.
  rwlock_wrunlock(&g_client_manager_rwlock);
 
  // Phase 2: Stop all client threads
  // For TCP clients: use tcp_server thread pool management
  // For WebRTC clients: manually join threads (no socket-based thread pool)
  // Use is_tcp_client flag, not socket value - socket may already be INVALID_SOCKET_VALUE
  // even for TCP clients if it was closed earlier during cleanup.
  bool receive_thread_is_current = false;
  log_debug("Stopping all threads for client %s (socket %d, is_tcp=%d)", client_id, client_socket,
            target_client ? target_client->is_tcp_client : -1);
 
  if (target_client && target_client->is_tcp_client) {
    // TCP client: use tcp_server thread pool
    // This joins threads in stop_id order: receive(1), render(2), send(3)
    // Use saved client_socket for lookup (tcp_server needs original socket as key)
    if (client_socket != INVALID_SOCKET_VALUE) {
      asciichat_error_t stop_result = tcp_server_stop_client_threads(server_ctx->tcp_server, client_socket);
      if (stop_result != ASCIICHAT_OK) {
        log_warn("Failed to stop threads for TCP client %s: error %d", client_id, stop_result);
        // Continue with cleanup even if thread stopping failed
      }
    } else {
      log_debug("TCP client %s socket already closed, threads should have already exited", client_id);
    }
  } else if (target_client) {
    // WebRTC client: manually join threads
    log_debug("Stopping WebRTC client %s threads (receive and send)", client_id);
 
    // Join receive thread (but skip if called from the receive thread itself to avoid deadlock)
    thread_id_t current_thread_id = asciichat_thread_self();
    receive_thread_is_current = asciichat_thread_equal(current_thread_id, target_client->receive_thread_id);
    if (receive_thread_is_current) {
      log_debug("remove_client() called from receive thread for client %s, skipping self-join", client_id);
    } else {
      void *recv_result = NULL;
      asciichat_error_t recv_join_result = asciichat_thread_join(&target_client->receive_thread, &recv_result);
      if (recv_join_result != ASCIICHAT_OK) {
        log_warn("Failed to join receive thread for WebRTC client %s: error %d", client_id, recv_join_result);
      } else {
        log_debug("Joined receive thread for WebRTC client %s", client_id);
      }
    }
 
    // Join dispatch thread (BEFORE destroying packet queues)
    if (asciichat_thread_is_initialized(&target_client->dispatch_thread)) {
      atomic_store_bool(&target_client->dispatch_thread_running, false);
      void *dispatch_result = NULL;
      asciichat_error_t dispatch_join_result = asciichat_thread_join(&target_client->dispatch_thread, &dispatch_result);
      if (dispatch_join_result != ASCIICHAT_OK) {
        log_warn("Failed to join dispatch thread for WebRTC client %s: error %d", client_id, dispatch_join_result);
      } else {
        log_debug("Joined dispatch thread for WebRTC client %s", client_id);
      }
    }
 
    // Join send thread
    void *send_result = NULL;
    asciichat_error_t send_join_result = asciichat_thread_join(&target_client->send_thread, &send_result);
    if (send_join_result != ASCIICHAT_OK) {
      log_warn("Failed to join send thread for WebRTC client %s: error %d", client_id, send_join_result);
    } else {
      log_debug("Joined send thread for WebRTC client %s", client_id);
    }
 
    // Render workers access the client buffers and synchronization primitives.
    // Join them before those resources are destroyed or the client slot is reused.
    stop_client_render_threads(target_client);
  }
 
  // Destroy ACIP transport before closing socket
  // For WebSocket clients: LWS_CALLBACK_CLOSED already closed and destroyed the transport
  // Trying to destroy it again causes heap-use-after-free since LWS callbacks might still fire
  // For TCP clients: transport is ours to clean up
  if (target_client && target_client->transport && target_client->is_tcp_client) {
    acip_transport_destroy(target_client->transport);
    target_client->transport = NULL;
    log_debug("Destroyed ACIP transport for TCP client %s", client_id);
  } else if (target_client && target_client->transport && !target_client->is_tcp_client) {
    // WebSocket client - just NULL it out, LWS_CALLBACK_CLOSED already destroyed it
    target_client->transport = NULL;
    log_debug("Skipped transport destruction for WebSocket client %s (LWS already destroyed)", client_id);
  }
 
  // Now safe to close the socket (threads are stopped)
  if (client_socket != INVALID_SOCKET_VALUE) {
    log_debug("SOCKET_DEBUG: Closing socket %d for client %s after thread cleanup", client_socket, client_id);
    socket_close(client_socket);
  }
 
  // Phase 3: Clean up resources with write lock
  rwlock_wrlock(&g_client_manager_rwlock);
 
  // Re-validate target_client pointer after reacquiring lock.
  // Another thread might have invalidated the pointer while we had the lock released.
  if (target_client) {
    // Verify client_id still matches and client is still in shutting_down state
    bool still_shutting_down = atomic_load_bool(&target_client->shutting_down);
    if (strcmp(target_client->client_id, client_id) != 0 || !still_shutting_down) {
      log_warn("Client %s pointer invalidated during thread cleanup (id=%s, shutting_down=%d)", client_id,
               target_client->client_id, still_shutting_down);
      rwlock_wrunlock(&g_client_manager_rwlock);
      return 0; // Another thread completed the cleanup
    }
  }
 
  // Mark socket as closed in client structure
  if (target_client && target_client->socket != INVALID_SOCKET_VALUE) {
    mutex_lock(&target_client->client_state_mutex);
    target_client->socket = INVALID_SOCKET_VALUE;
    mutex_unlock(&target_client->client_state_mutex);
    log_debug("SOCKET_DEBUG: Client %s socket set to INVALID", target_client->client_id);
  }
 
  // Use the dedicated cleanup function to ensure all resources are freed
  cleanup_client_all_buffers(target_client);
 
  // Remove from audio mixer
  if (g_audio_mixer) {
    mixer_remove_source(g_audio_mixer, client_id);
#ifdef DEBUG_AUDIO
    log_debug("Removed client %s from audio mixer", client_id);
#endif
  }
 
  // Remove from uthash table
  // Verify client is actually in the hash table before deleting.
  // Another thread might have already removed it.
  if (target_client) {
    client_info_t *hash_entry = NULL;
    HASH_FIND_STR(g_client_manager.clients_by_id, target_client->client_id, hash_entry);
    if (hash_entry == target_client) {
      HASH_DELETE(hh, g_client_manager.clients_by_id, target_client);
      log_debug("Removed client %s from uthash table", client_id);
    } else {
      log_warn("Client %s already removed from hash table by another thread (found=%p, expected=%p)", client_id,
               (void *)hash_entry, (void *)target_client);
    }
  } else {
    log_warn("Failed to remove client %s from hash table (client not found)", client_id);
  }
 
  // Cleanup crypto context for this client
  if (target_client->crypto_initialized) {
    crypto_handshake_destroy(&target_client->crypto_handshake_ctx);
    target_client->crypto_initialized = false;
    log_debug("Crypto context cleaned up for client %s", client_id);
  }
 
  // Verify all threads have actually exited before resetting client_id.
  // Threads that are still starting (at RtlUserThreadStart) haven't checked client_id yet.
  // We must ensure threads are fully joined before zeroing the client struct.
  // Use exponential backoff for thread termination verification
  int retry_count = 0;
  const int max_retries = 5;
  while (retry_count < max_retries && (asciichat_thread_is_initialized(&target_client->send_thread) ||
                                       (!receive_thread_is_current &&
                                        asciichat_thread_is_initialized(&target_client->receive_thread)) ||
                                       asciichat_thread_is_initialized(&target_client->video_render_thread) ||
                                       asciichat_thread_is_initialized(&target_client->audio_render_thread))) {
    // Exponential backoff: 10ms, 20ms, 40ms, 80ms, 160ms
    uint32_t delay_ms = 10 * (1 << retry_count);
    log_warn("Client %s: Some threads still appear initialized (attempt %d/%d), waiting %ums", client_id,
             retry_count + 1, max_retries, delay_ms);
    platform_sleep_us(delay_ms * 1000);
    retry_count++;
  }
 
  if (retry_count == max_retries) {
    log_error("Client %s: Threads did not terminate after %d retries, proceeding with cleanup anyway", client_id,
              max_retries);
  }
 
  // Only reset client_id to 0 AFTER confirming threads are joined
  // This prevents threads that are starting from accessing a zeroed client struct
  // Reset client_id to NULL before destroying mutexes to prevent race conditions.
  // This ensures worker threads can detect shutdown and exit before the mutex is destroyed.
  // If we destroy the mutex first, threads might try to access a destroyed mutex.
  target_client->client_id[0] = '\0'; // Clear the client_id string
 
  // Wait for threads to observe the client_id reset
  // Use sufficient delay for memory visibility across all CPU cores
  platform_sleep_us(5 * US_PER_MS_INT); // 5ms delay for memory barrier propagation
 
  // Destroy mutexes and condition variables
  // IMPORTANT: Always destroy these even if threads didn't join properly
  // to prevent issues when the slot is reused
  mutex_destroy(&target_client->client_state_mutex);
  mutex_destroy(&target_client->send_mutex);
  cond_destroy(&target_client->dispatch_queue_cond);
 
  // Clear client structure
  // NOTE: After memset, the mutex handles are zeroed but the OS resources
  // have been released by the destroy calls above
  memset(target_client, 0, sizeof(client_info_t));
 
  // Recalculate client count
  int remaining_count = 0;
  for (int j = 0; j < MAX_CLIENTS; j++) {
    if (g_client_manager.clients[j].client_id[0] != '\0') {
      remaining_count++;
    }
  }
  g_client_manager.client_count = remaining_count;
 
  log_debug("Client removed: client_id=%s (%s) removed, remaining clients: %d", client_id, display_name_copy,
            remaining_count);
 
  rwlock_wrunlock(&g_client_manager_rwlock);
 
  // Broadcast updated state
  broadcast_server_state_to_all_clients();
 
  return 0;
}

References acip_transport_destroy(), client_info::active, ASCIICHAT_OK, asciichat_thread_equal, asciichat_thread_is_initialized(), asciichat_thread_join, asciichat_thread_self, atomic_load_bool(), atomic_store_bool(), client_info::audio_queue, client_info::audio_render_thread, broadcast_server_state_to_all_clients(), client_manager_t::client_count, client_info::client_id, client_info::client_state_mutex, client_manager_t::clients, client_manager_t::clients_by_id, cond_destroy(), client_info::crypto_handshake_ctx, crypto_handshake_destroy(), client_info::crypto_initialized, client_info::dispatch_queue_cond, client_info::dispatch_thread, client_info::dispatch_thread_running, client_info::display_name, ERROR_INVALID_PARAM, ERROR_NOT_FOUND, g_audio_mixer, g_client_manager, g_client_manager_rwlock, INVALID_SOCKET_VALUE, client_info::is_sending_audio, client_info::is_sending_video, client_info::is_tcp_client, log_debug, log_error, log_info, log_warn, MAX_CLIENTS, MAX_DISPLAY_NAME_LEN, mixer_remove_source(), mutex_destroy(), mutex_lock, mutex_unlock, packet_queue_stop(), platform_sleep_us(), client_info::receive_thread, client_info::receive_thread_id, rwlock_wrlock, rwlock_wrunlock, SAFE_STRNCPY, client_info::send_mutex, client_info::send_thread, client_info::session_client_id, server_context_t::session_host, session_host_remove_client(), SET_ERRNO, client_info::shutting_down, client_info::socket, socket_close(), socket_shutdown(), stop_client_render_threads(), server_context_t::tcp_server, tcp_server_stop_client_threads(), client_info::transport, US_PER_MS_INT, and client_info::video_render_thread.

Referenced by add_client(), add_webrtc_client(), client_receive_thread(), and client_send_thread_func().

start_webrtc_client_threads()

int start_webrtc_client_threads	(	server_context_t *	server_ctx,
		const char *	client_id
	)

Start threads for a WebRTC client after crypto initialization.

This is called by WebSocket handler after crypto handshake is initialized. It ensures receive thread doesn't try to process packets before crypto context exists.

Parameters

server_ctx	Server context
client_id	Client ID to start threads for

Returns

0 on success, -1 on failure

Definition at line 2832 of file src/server/client.c.

                                                                                     {
  if (!server_ctx) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Server context is NULL");
    return -1;
  }
 
  client_info_t *client = find_client_by_id(client_id);
  if (!client) {
    SET_ERRNO(ERROR_NOT_FOUND, "Client %s not found", client_id);
    return -1;
  }
 
  log_debug("Starting threads for WebRTC client %s...", client_id);
  return start_client_threads(server_ctx, client, false);
}

References ERROR_INVALID_PARAM, ERROR_NOT_FOUND, find_client_by_id(), log_debug, and SET_ERRNO.

stop_client_threads()

void stop_client_threads

(

client_info_t *

client

)

Definition at line 2848 of file src/server/client.c.

                                                {
  if (!client) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Client is NULL");
    return;
  }
 
  // Signal threads to stop
  log_debug("Setting active=false in stop_client_threads (client_id=%s)", client->client_id);
  atomic_store_bool(&client->active, false);
  atomic_store_bool(&client->send_thread_running, false);
 
  // Wait for threads to finish
  if (asciichat_thread_is_initialized(&client->send_thread)) {
    asciichat_thread_join(&client->send_thread, NULL);
  }
  if (asciichat_thread_is_initialized(&client->receive_thread)) {
    asciichat_thread_join(&client->receive_thread, NULL);
  }
  // For async dispatch: stop dispatch thread if running
  if (asciichat_thread_is_initialized(&client->dispatch_thread)) {
    atomic_store_bool(&client->dispatch_thread_running, false);
    asciichat_thread_join(&client->dispatch_thread, NULL);
  }
}

References client_info::active, asciichat_thread_is_initialized(), asciichat_thread_join, atomic_store_bool(), client_info::client_id, client_info::dispatch_thread, client_info::dispatch_thread_running, ERROR_INVALID_PARAM, log_debug, client_info::receive_thread, client_info::send_thread, client_info::send_thread_running, and SET_ERRNO.

Variable Documentation

g_client_manager

client_manager_t g_client_manager

Global client manager singleton - central coordination point.

Global client manager.

This is the primary data structure for managing all connected clients. It serves as the bridge between main.c's connection accept loop and the per-client threading architecture.

STRUCTURE COMPONENTS:

• clients[]: Array backing storage for client_info_t structs
• client_hashtable: O(1) lookup table for client_id -> client_info_t*
• client_count: Current number of active clients
• mutex: Legacy mutex (mostly replaced by rwlock)
• next_client_id: Monotonic counter for unique client identification

THREAD SAFETY: Protected by g_client_manager_rwlock for concurrent access

Definition at line 347 of file src/server/client.c.

Referenced by add_client(), add_webrtc_client(), any_clients_sending_video(), broadcast_server_state_to_all_clients(), find_client_by_id(), find_client_by_socket(), remove_client(), send_server_state_to_client(), server_main(), stats_logger_thread(), and update_server_stats().

g_client_manager_rwlock

rwlock_t g_client_manager_rwlock = {0}

Reader-writer lock protecting the global client manager.

Read-write lock protecting client manager.

This lock enables high-performance concurrent access patterns:

• Multiple threads can read client data simultaneously (stats, rendering)
• Only one thread can modify client data at a time (add/remove operations)
• Eliminates contention between read-heavy operations

USAGE PATTERN:

• Read operations: rwlock_rdlock() for client lookups, stats gathering
• Write operations: rwlock_wrlock() for add_client(), remove_client()
• Always acquire THIS lock before per-client mutexes (lock ordering)

Definition at line 362 of file src/server/client.c.

{0};

Referenced by add_client(), add_webrtc_client(), broadcast_server_state_to_all_clients(), find_client_by_id(), find_client_by_socket(), remove_client(), server_main(), stats_logger_thread(), and update_server_stats().

g_client_manager_rwlock_initialized

bool g_client_manager_rwlock_initialized = false

Track whether g_client_manager_rwlock was successfully initialized.

Definition at line 365 of file src/server/client.c.