client.h File Reference

Per-client state management and lifecycle orchestration. More...

Go to the source code of this file.

Data Structures
struct	client_manager_t
	Global client manager structure for server-side client coordination. More...

Typedefs
typedef struct server_context_t	server_context_t

Functions
int	add_client (server_context_t *server_ctx, socket_t socket, const char *client_ip, int port)
int	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	start_webrtc_client_threads (server_context_t *server_ctx, uint32_t client_id)
	Start threads for a WebRTC client after crypto initialization.
int	remove_client (server_context_t *server_ctx, uint32_t client_id)
client_info_t *	find_client_by_id (uint32_t 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.
void	cleanup_client_media_buffers (client_info_t *client)
void	cleanup_client_packet_queues (client_info_t *client)
void *	client_receive_thread (void *arg)
void	stop_client_threads (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)
void	initialize_client_info (client_info_t *client)

Variables
client_manager_t	g_client_manager
	Global client manager singleton - central coordination point.
rwlock_t	g_client_manager_rwlock
	Reader-writer lock protecting the global client manager.

Detailed Description

Per-client state management and lifecycle orchestration.

This header provides server-specific client management functions. The client_info_t structure and network logging macros are defined in lib/network/client.h.

Definition in file client.h.

Typedef Documentation

server_context_t

typedef struct server_context_t server_context_t

Definition at line 16 of file client.h.

Function Documentation

add_client()

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

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

                                                                                               {
  // 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 && atomic_load(&g_client_manager.clients[i].client_id) == 0) {
      slot = i; // Take first available slot
    }
    if (atomic_load(&g_client_manager.clients[i].client_id) != 0 && atomic_load(&g_client_manager.clients[i].active)) {
      existing_count++;
    }
  }
 
  // Quick pre-check before expensive allocations
  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 -1;
  }
 
  // DO EXPENSIVE ALLOCATIONS OUTSIDE THE LOCK
  // This prevents blocking frame processing during client initialization
  uint32_t new_client_id = atomic_fetch_add(&g_client_manager.next_client_id, 1) + 1;
 
  video_frame_buffer_t *incoming_video_buffer = video_frame_buffer_create(new_client_id);
  if (!incoming_video_buffer) {
    SET_ERRNO(ERROR_MEMORY, "Failed to create video buffer for client %u", new_client_id);
    log_error("Failed to create video buffer for client %u", new_client_id);
    return -1;
  }
 
  audio_ring_buffer_t *incoming_audio_buffer = audio_ring_buffer_create_for_capture();
  if (!incoming_audio_buffer) {
    SET_ERRNO(ERROR_MEMORY, "Failed to create audio buffer for client %u", new_client_id);
    log_error("Failed to create audio buffer for client %u", new_client_id);
    video_frame_buffer_destroy(incoming_video_buffer);
    return -1;
  }
 
  packet_queue_t *audio_queue = packet_queue_create_with_pools(500, 1000, false);
  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 -1;
  }
 
  video_frame_buffer_t *outgoing_video_buffer = video_frame_buffer_create(new_client_id);
  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 -1;
  }
 
  void *send_buffer = SAFE_MALLOC_ALIGNED(MAX_FRAME_BUFFER_SIZE, 64, void *);
  if (!send_buffer) {
    log_error("Failed to allocate send buffer for client %u", 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 -1;
  }
 
  // NOW acquire the lock only for the critical section: slot assignment + registration
  rwlock_wrlock(&g_client_manager_rwlock);
 
  // Re-check slot availability under lock (another thread might have taken it)
  if (atomic_load(&g_client_manager.clients[slot].client_id) != 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 -1;
  }
 
  // 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));
 
  client->socket = socket;
  client->is_tcp_client = true;
  atomic_store(&client->client_id, new_client_id);
  SAFE_STRNCPY(client->client_ip, client_ip, sizeof(client->client_ip) - 1);
  client->port = port;
  atomic_store(&client->active, true);
  client->server_ctx = server_ctx; // Store server context for cleanup
  atomic_store(&client->shutting_down, false);
  atomic_store(&client->last_rendered_grid_sources, 0);
  atomic_store(&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->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;
 
  safe_snprintf(client->display_name, sizeof(client->display_name), "Client%u", new_client_id);
  log_info("Added new client ID=%u from %s:%d (socket=%d, slot=%d)", new_client_id, client_ip, port, socket, slot);
  log_debug("Client slot assigned: client_id=%u 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 %u 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=%u to slot %d)", g_client_manager.client_count,
            new_client_id, slot);
 
  uint32_t cid = atomic_load(&client->client_id);
  HASH_ADD_INT(g_client_manager.clients_by_id, client_id, client);
  log_debug("Added client %u to uthash table", cid);
 
  rwlock_wrunlock(&g_client_manager_rwlock);
 
  // Configure socket OUTSIDE lock
  configure_client_socket(socket, new_client_id);
 
  // Initialize mutexes OUTSIDE lock
  if (mutex_init(&client->client_state_mutex) != 0) {
    log_error("Failed to initialize client state mutex for client %u", new_client_id);
    remove_client(server_ctx, new_client_id);
    return -1;
  }
 
  if (mutex_init(&client->send_mutex) != 0) {
    log_error("Failed to initialize send mutex for client %u", new_client_id);
    remove_client(server_ctx, new_client_id);
    return -1;
  }
 
  // Register with audio mixer OUTSIDE lock
  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 client %u to audio mixer", new_client_id);
    } else {
#ifdef DEBUG_AUDIO
      log_debug("Added client %u to audio mixer", new_client_id);
#endif
    }
  }
 
  // Perform crypto handshake before starting threads.
  // This ensures the handshake uses the socket directly without interference from receive thread.
  if (server_crypto_init() == 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 %u: %s", atomic_load(&client->client_id),
               asciichat_error_string(timeout_result));
      // Continue anyway - timeout is a safety feature, not critical
    }
 
    int crypto_result = server_crypto_handshake(client);
    if (crypto_result != 0) {
      log_error("Crypto handshake failed for client %u: %s", atomic_load(&client->client_id), network_error_string());
      if (remove_client(server_ctx, atomic_load(&client->client_id)) != 0) {
        log_error("Failed to remove client after crypto handshake failure");
      }
      return -1;
    }
 
    // Clear socket timeout after handshake completes successfully
    // This allows normal operation without timeouts on data transfer
    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 %u: %s", atomic_load(&client->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 %u", atomic_load(&client->client_id));
 
    // Create ACIP transport for protocol-agnostic packet sending
    // The transport wraps the socket with encryption context from the handshake
    const crypto_context_t *crypto_ctx = crypto_server_get_context(atomic_load(&client->client_id));
    client->transport = acip_tcp_transport_create(socket, (crypto_context_t *)crypto_ctx);
    if (!client->transport) {
      log_error("Failed to create ACIP transport for client %u", atomic_load(&client->client_id));
      if (remove_client(server_ctx, atomic_load(&client->client_id)) != 0) {
        log_error("Failed to remove client after transport creation failure");
      }
      return -1;
    }
    log_debug("Created ACIP transport for client %u with crypto context", atomic_load(&client->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 %u using --no-encrypt mode - processing pending packet type %u", atomic_load(&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 %u", atomic_load(&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(atomic_load(&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 %u: result=%d",
                  atomic_load(&client->client_id), result);
        if (envelope.allocated_buffer) {
          buffer_pool_free(NULL, envelope.allocated_buffer, envelope.allocated_size);
        }
        if (remove_client(server_ctx, atomic_load(&client->client_id)) != 0) {
          log_error("Failed to remove client after crypto handshake failure");
        }
        return -1;
      }
    }
 
    if (envelope.type != PACKET_TYPE_CLIENT_CAPABILITIES) {
      log_error("Expected PACKET_TYPE_CLIENT_CAPABILITIES but got packet type %d from client %u", envelope.type,
                atomic_load(&client->client_id));
      if (envelope.allocated_buffer) {
        buffer_pool_free(NULL, envelope.allocated_buffer, envelope.allocated_size);
      }
      if (remove_client(server_ctx, atomic_load(&client->client_id)) != 0) {
        log_error("Failed to remove client after crypto handshake failure");
      }
      return -1;
    }
 
    // Process the capabilities packet directly
    log_debug("Processing initial capabilities packet from client %u (from %s)", atomic_load(&client->client_id),
              used_pending_packet ? "pending packet" : "network");
    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 %u",
              atomic_load(&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
  uint32_t client_id_snapshot = atomic_load(&client->client_id);
  if (start_client_threads(server_ctx, client, true) != 0) {
    log_error("Failed to start threads for TCP client %u", client_id_snapshot);
    // Client is already in hash table - use remove_client for proper cleanup
    remove_client(server_ctx, client_id_snapshot);
    return -1;
  }
  log_debug("Successfully created render threads for client %u", client_id_snapshot);
 
  // Register client with session_host (for discovery mode support)
  if (server_ctx->session_host) {
    uint32_t session_client_id = session_host_add_client(server_ctx->session_host, socket, client_ip, port);
    if (session_client_id == 0) {
      log_warn("Failed to register client %u with session_host", client_id_snapshot);
    } else {
      log_debug("Client %u registered with session_host as %u", client_id_snapshot, 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();
 
  return (int)client_id_snapshot;
 
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 -1;
}

References acip_tcp_transport_create(), 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_manager_t::clients, client_manager_t::clients_by_id, crypto_handshake_is_ready(), crypto_server_get_context(), g_audio_mixer, g_client_manager, g_client_manager_rwlock, handle_client_capabilities_packet(), mixer_add_source(), mutex_init(), network_error_string(), client_manager_t::next_client_id, packet_queue_create_with_pools(), packet_queue_destroy(), receive_packet_secure(), remove_client(), safe_snprintf(), server_crypto_handshake(), server_crypto_init(), server_context_t::session_host, session_host_add_client(), set_socket_timeout(), server_context_t::tcp_server, tcp_server_add_client(), video_frame_buffer_create(), and video_frame_buffer_destroy().

add_webrtc_client()

int 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

Client ID on success, -1 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 887 of file src/server/client.c.

                                          {
  if (!server_ctx || !transport || !client_ip) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameters to add_webrtc_client");
    return -1;
  }
 
  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 && atomic_load(&g_client_manager.clients[i].client_id) == 0) {
      slot = i; // Take first available slot
    }
    // Count only active clients
    if (atomic_load(&g_client_manager.clients[i].client_id) != 0 && atomic_load(&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 -1;
  }
 
  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 -1;
  }
 
  // Update client_count to match actual count before adding new client
  g_client_manager.client_count = existing_count;
 
  // 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 = transport;         // Use provided transport
  uint32_t new_client_id = atomic_fetch_add(&g_client_manager.next_client_id, 1) + 1;
  atomic_store(&client->client_id, new_client_id);
  SAFE_STRNCPY(client->client_ip, client_ip, sizeof(client->client_ip) - 1);
  client->port = 0; // WebRTC doesn't use port numbers
  atomic_store(&client->active, true);
  client->server_ctx = server_ctx; // Store server context for receive thread cleanup
  log_info("Added new WebRTC client ID=%u from %s (transport=%p, slot=%d)", new_client_id, client_ip, transport, slot);
  atomic_store(&client->shutting_down, false);
  atomic_store(&client->last_rendered_grid_sources, 0); // Render thread updates this
  atomic_store(&client->last_sent_grid_sources, 0);     // Send thread updates this
  log_debug("WebRTC client slot assigned: client_id=%u assigned to slot %d", atomic_load(&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;
 
  safe_snprintf(client->display_name, sizeof(client->display_name), "WebRTC%u", atomic_load(&client->client_id));
 
  // Create individual video buffer for this client using modern double-buffering
  client->incoming_video_buffer = video_frame_buffer_create(atomic_load(&client->client_id));
  if (!client->incoming_video_buffer) {
    SET_ERRNO(ERROR_MEMORY, "Failed to create video buffer for WebRTC client %u", atomic_load(&client->client_id));
    log_error("Failed to create video buffer for WebRTC client %u", atomic_load(&client->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 %u", atomic_load(&client->client_id));
    log_error("Failed to create audio buffer for WebRTC client %u", atomic_load(&client->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(atomic_load(&client->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 %u", atomic_load(&client->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=%u to slot %d)",
            g_client_manager.client_count, atomic_load(&client->client_id), slot);
 
  // Add client to uthash table for O(1) lookup
  uint32_t cid = atomic_load(&client->client_id);
  HASH_ADD_INT(g_client_manager.clients_by_id, client_id, client);
  log_debug("Added WebRTC client %u to uthash table", cid);
 
  // Register this client's audio buffer with the mixer
  if (g_audio_mixer && client->incoming_audio_buffer) {
    if (mixer_add_source(g_audio_mixer, atomic_load(&client->client_id), client->incoming_audio_buffer) < 0) {
      log_warn("Failed to add WebRTC client %u to audio mixer", atomic_load(&client->client_id));
    } else {
#ifdef DEBUG_AUDIO
      log_debug("Added WebRTC client %u to audio mixer", atomic_load(&client->client_id));
#endif
    }
  }
 
  // Initialize mutexes BEFORE creating any threads to prevent race conditions
  if (mutex_init(&client->client_state_mutex) != 0) {
    log_error("Failed to initialize client state mutex for WebRTC client %u", atomic_load(&client->client_id));
    // Client is already in hash table - use remove_client for proper cleanup
    rwlock_wrunlock(&g_client_manager_rwlock);
    remove_client(server_ctx, atomic_load(&client->client_id));
    return -1;
  }
 
  // Initialize send mutex to protect concurrent socket writes
  if (mutex_init(&client->send_mutex) != 0) {
    log_error("Failed to initialize send mutex for WebRTC client %u", atomic_load(&client->client_id));
    // Client is already in hash table - use remove_client for proper cleanup
    rwlock_wrunlock(&g_client_manager_rwlock);
    remove_client(server_ctx, atomic_load(&client->client_id));
    return -1;
  }
 
  rwlock_wrunlock(&g_client_manager_rwlock);
 
  // 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 %u initialized - receive thread will process capabilities", atomic_load(&client->client_id));
 
  uint32_t client_id_snapshot = atomic_load(&client->client_id);
 
  // Conditionally start threads based on caller preference
  // WebSocket handler passes start_threads=false to defer thread startup until after crypto init
  // This ensures receive thread doesn't try to process packets before crypto context is ready
  if (start_threads) {
    log_debug("[ADD_WEBRTC_CLIENT] Starting client threads (receive, render) for client %u...", client_id_snapshot);
    if (start_client_threads(server_ctx, client, false) != 0) {
      log_error("Failed to start threads for WebRTC client %u", client_id_snapshot);
      return -1;
    }
    log_debug("Created receive thread for WebRTC client %u", client_id_snapshot);
    log_debug("[ADD_WEBRTC_CLIENT] Receive thread started - thread will now be processing packets", client_id_snapshot);
  } else {
    log_debug("[ADD_WEBRTC_CLIENT] Deferring thread startup for client %u (caller will start after crypto init)",
              client_id_snapshot);
  }
 
  // 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 %u", client_id_snapshot);
  } else {
#ifdef DEBUG_NETWORK
    log_info("Sent initial server state to WebRTC client %u", client_id_snapshot);
#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 %u: %s", client_id_snapshot,
             asciichat_error_string(packet_send_result));
  } else {
    log_debug("Sent initial server state to WebRTC client %u: %u connected clients", client_id_snapshot,
              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) {
    uint32_t session_client_id = session_host_add_client(server_ctx->session_host, INVALID_SOCKET_VALUE, client_ip, 0);
    if (session_client_id == 0) {
      log_warn("Failed to register WebRTC client %u with session_host", client_id_snapshot);
    } else {
      log_debug("WebRTC client %u registered with session_host as %u", client_id_snapshot, 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();
 
  return (int)client_id_snapshot;
 
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 -1;
}

References acip_send_server_state(), audio_ring_buffer_create_for_capture(), audio_ring_buffer_destroy(), broadcast_server_state_to_all_clients(), client_manager_t::client_count, client_manager_t::clients, client_manager_t::clients_by_id, g_audio_mixer, g_client_manager, g_client_manager_rwlock, mixer_add_source(), mutex_init(), client_manager_t::next_client_id, packet_queue_create_with_pools(), remove_client(), safe_snprintf(), send_server_state_to_client(), server_context_t::session_host, session_host_add_client(), video_frame_buffer_create(), and video_frame_buffer_destroy().

cleanup_client_media_buffers()

void cleanup_client_media_buffers

(

client_info_t *

client

)

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

                                                         {
  if (!client) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Client is NULL");
    return;
  }
 
  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;
  }
}

References audio_ring_buffer_destroy(), opus_codec_destroy(), and video_frame_buffer_destroy().

cleanup_client_packet_queues()

void cleanup_client_packet_queues

(

client_info_t *

client

)

Definition at line 2474 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 packet_queue_destroy().

client_receive_thread()

void * client_receive_thread

(

void *

arg

)

Definition at line 1582 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=%u, is_tcp=%d", (void *)client,
            atomic_load(&client->client_id), client->is_tcp_client);
 
  if (atomic_load(&client->protocol_disconnect_requested)) {
    log_debug("Receive thread for client %u exiting before start (protocol disconnect requested)",
              atomic_load(&client->client_id));
    return NULL;
  }
 
  // Check if client_id is 0 (client struct has been zeroed by remove_client)
  // This must be checked BEFORE accessing any client fields
  if (atomic_load(&client->client_id) == 0) {
    log_debug("Receive thread: client_id is 0, 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;
  }
 
  // Enable thread cancellation for clean shutdown
  // Thread cancellation not available in platform abstraction
  // Threads should exit when g_server_should_exit is set
 
  log_debug("Started receive thread for client %u (%s)", atomic_load(&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=%u, is_tcp=%d, transport=%p, active=%d", atomic_load(&client->client_id),
           client->is_tcp_client, (void *)client->transport, atomic_load(&client->active));
 
  while (!atomic_load(&g_server_should_exit) && atomic_load(&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 %u has invalid socket, exiting receive thread", atomic_load(&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 (atomic_load(&client->client_id) == 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(&g_server_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 %u: code=%u msg=%s", client->client_id, err_ctx.code,
                    err_ctx.context_message);
          if (err_ctx.code == ERROR_NETWORK) {
            log_debug("Client %u 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(&g_server_should_exit, true);
            break;
          }
        }
        log_warn("ACIP error for TCP client %u: %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;
 
      asciichat_error_t recv_result =
          client->transport->methods->recv(client->transport, &packet_data, &packet_len, &allocated_buffer);
 
      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 %u: fragment reassembly timeout, retrying in 10ms", client->client_id);
            platform_sleep_ms(10); // Sleep 10ms to allow fragments to arrive
            continue;             // Retry without disconnecting
          }
 
          if (err_ctx.code == ERROR_NETWORK) {
            log_debug("Client %u disconnected (network error): %s", client->client_id, err_ctx.context_message);
            break;
          }
        }
        log_warn("Receive failed for WebRTC client %u: %s (disconnecting)", client->client_id,
                 asciichat_error_string(recv_result));
        break;
      }
 
      // Queue the received packet for async dispatch
      // This prevents the receive thread from blocking on dispatch
      log_info("RECV_THREAD: Queuing %zu byte packet for async dispatch (client %u)", packet_len, client->client_id);
 
      // Extract packet type from the header to preserve it when queueing
      packet_type_t pkt_type = PACKET_TYPE_PROTOCOL_VERSION;
      if (packet_len >= sizeof(packet_header_t)) {
        const packet_header_t *pkt_header = (const packet_header_t *)allocated_buffer;
        pkt_type = (packet_type_t)NET_TO_HOST_U16(pkt_header->type);
      }
 
      // Build a complete packet to queue (header + payload)
      // The entire buffer (allocated_buffer) contains the full packet
      // copy_data=false because we want to transfer ownership to the queue
      int enqueue_result = packet_queue_enqueue(client->received_packet_queue, pkt_type, allocated_buffer, packet_len,
                                                atomic_load(&client->client_id), false);
 
      if (enqueue_result < 0) {
        log_warn("Failed to queue received packet for client %u (queue full?) - packet dropped", client->client_id);
        if (allocated_buffer) {
          buffer_pool_free(NULL, allocated_buffer, 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)
  uint32_t client_id_snapshot = atomic_load(&client->client_id);
  log_debug("Setting active=false in receive_thread_fn (client_id=%u, exiting receive loop)", client_id_snapshot);
  atomic_store(&client->active, false);
  atomic_store(&client->send_thread_running, false);
  atomic_store(&client->video_render_thread_running, false);
  atomic_store(&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 %u calling remove_client() for cleanup", client_id_snapshot);
  server_context_t *server_ctx = (server_context_t *)client->server_ctx;
  if (server_ctx) {
    if (remove_client(server_ctx, client_id_snapshot) != 0) {
      log_warn("Failed to remove client %u from receive thread cleanup", client_id_snapshot);
    }
  } else {
    log_error("Receive thread for client %u: server_ctx is NULL, cannot call remove_client()", client_id_snapshot);
  }
 
  log_debug("Receive thread for client %u terminated", client_id_snapshot);
 
  // Clean up thread-local error context before exit
  asciichat_errno_destroy();
 
  return NULL;
}

References acip_server_receive_and_dispatch(), asciichat_errno_destroy(), asciichat_thread_self(), buffer_pool_free(), g_server_should_exit, packet_queue_enqueue(), platform_sleep_ms(), and remove_client().

find_client_by_id()

client_info_t * find_client_by_id

(

uint32_t

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 308 of file src/server/client.c.

                                                     {
  if (client_id == 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;
  uint32_t search_id = client_id; // uthash needs an lvalue for the key
  HASH_FIND_INT(g_client_manager.clients_by_id, &search_id, result);
 
  rwlock_rdunlock(&g_client_manager_rwlock);
 
  if (!result) {
    log_warn("Client not found for ID %u", client_id);
  }
 
  return result;
}

References client_manager_t::clients_by_id, g_client_manager, and g_client_manager_rwlock.

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 353 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(&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_manager_t::clients, g_client_manager, and g_client_manager_rwlock.

initialize_client_info()

void initialize_client_info

(

client_info_t *

client

)

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 3136 of file src/server/client.c.

                                                                                                 {
  if (type == 5000) { // CLIENT_CAPABILITIES
    log_debug("CLIENT: client_id=%u, data=%p, len=%zu", atomic_load(&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(), disconnect_client_for_bad_data(), and g_rate_limiter.

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 2516 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 %u", 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 %u (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 %u", 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(), crypto_server_decrypt_packet(), and crypto_server_is_ready().

remove_client()

int remove_client	(	server_context_t *	server_ctx,
		uint32_t	client_id
	)

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

                                                                    {
  if (!server_ctx) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Cannot remove client %u: 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 %d", 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];
    uint32_t cid = atomic_load(&client->client_id);
    if (cid == client_id && cid != 0) {
      // Check if already being removed by another thread
      // This prevents double-free and use-after-free crashes during concurrent cleanup
      if (atomic_load(&client->shutting_down)) {
        rwlock_wrunlock(&g_client_manager_rwlock);
        log_debug("Client %u 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=%d, socket=%d)", client_id, client->socket);
      log_info("Removing client %d (socket=%d) - marking inactive and clearing video flags", client_id, client->socket);
      atomic_store(&client->shutting_down, true);
      atomic_store(&client->active, false);
      atomic_store(&client->is_sending_video, false);
      atomic_store(&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() before closing
      mutex_lock(&client->client_state_mutex);
      client_socket = client->socket; // Save socket for thread cleanup
      if (client->socket != INVALID_SOCKET_VALUE) {
        log_debug("SOCKET_DEBUG: Client %d shutting down socket %d", client->client_id, client->socket);
        // Shutdown both send and receive operations to unblock any pending I/O
        socket_shutdown(client->socket, 2); // 2 = SHUT_RDWR on POSIX, SD_BOTH on Windows
        // Don't close yet - tcp_server needs socket as lookup key
      }
      mutex_unlock(&client->client_state_mutex);
 
      // 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 %u: 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) {
    asciichat_error_t session_result = session_host_remove_client(server_ctx->session_host, 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 %u not found in session_host (likely failed crypto before registration)", client_id);
      } else {
        log_warn("Failed to unregister client %u from session_host: %s", client_id,
                 asciichat_error_string(session_result));
      }
    } else {
      log_debug("Client %u 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.
  log_debug("Stopping all threads for client %u (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 %u: error %d", client_id, stop_result);
        // Continue with cleanup even if thread stopping failed
      }
    } else {
      log_debug("TCP client %u socket already closed, threads should have already exited", client_id);
    }
  } else if (target_client) {
    // WebRTC client: manually join threads
    log_debug("Stopping WebRTC client %u 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();
    if (asciichat_thread_equal(current_thread_id, target_client->receive_thread_id)) {
      log_debug("remove_client() called from receive thread for client %u, 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 %u: error %d", client_id, recv_join_result);
      } else {
        log_debug("Joined receive thread for WebRTC client %u", 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 %u: error %d", client_id, send_join_result);
    } else {
      log_debug("Joined send thread for WebRTC client %u", client_id);
    }
    // Note: Render threads still need to be stopped - they're created the same way for both TCP and WebRTC
    // For now, render threads are expected to exit when they check g_server_should_exit and client->active
  }
 
  // 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 %u", 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 %u (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 %u 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
    uint32_t current_id = atomic_load(&target_client->client_id);
    bool still_shutting_down = atomic_load(&target_client->shutting_down);
    if (current_id != client_id || !still_shutting_down) {
      log_warn("Client %u pointer invalidated during thread cleanup (id=%u, shutting_down=%d)", client_id, current_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 %u socket set to INVALID", 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 %u 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(hh, g_client_manager.clients_by_id, &client_id, sizeof(client_id), hash_entry);
    if (hash_entry == target_client) {
      HASH_DELETE(hh, g_client_manager.clients_by_id, target_client);
      log_debug("Removed client %u from uthash table", client_id);
    } else {
      log_warn("Client %u 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 %u 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 %u", 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) ||
                                       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 %u: 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 %u: 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 0 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.
  atomic_store(&target_client->client_id, 0);
 
  // 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
  // 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);
 
  // 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 using atomic reads
  int remaining_count = 0;
  for (int j = 0; j < MAX_CLIENTS; j++) {
    if (atomic_load(&g_client_manager.clients[j].client_id) != 0) {
      remaining_count++;
    }
  }
  g_client_manager.client_count = remaining_count;
 
  log_debug("Client removed: client_id=%u (%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(), asciichat_thread_equal(), asciichat_thread_join(), asciichat_thread_self(), broadcast_server_state_to_all_clients(), client_manager_t::client_count, client_manager_t::clients, client_manager_t::clients_by_id, crypto_handshake_destroy(), g_audio_mixer, g_client_manager, g_client_manager_rwlock, mixer_remove_source(), mutex_destroy(), packet_queue_stop(), platform_sleep_us(), server_context_t::session_host, session_host_remove_client(), server_context_t::tcp_server, and tcp_server_stop_client_threads().

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

start_webrtc_client_threads()

int start_webrtc_client_threads	(	server_context_t *	server_ctx,
		uint32_t	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 2397 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 %u not found", client_id);
    return -1;
  }
 
  log_debug("Starting threads for WebRTC client %u...", client_id);
  return start_client_threads(server_ctx, client, false);
}

References find_client_by_id().

stop_client_threads()

void stop_client_threads

(

client_info_t *

client

)

Definition at line 2413 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=%u)", atomic_load(&client->client_id));
  atomic_store(&client->active, false);
  atomic_store(&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(&client->dispatch_thread_running, false);
    asciichat_thread_join(&client->dispatch_thread, NULL);
  }
}

References asciichat_thread_join().

Variable Documentation

g_client_manager

client_manager_t g_client_manager

extern

Global client manager singleton - central coordination point.

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 255 of file src/server/client.c.

g_client_manager_rwlock

rwlock_t g_client_manager_rwlock

extern

Reader-writer lock protecting the global 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 270 of file src/server/client.c.

{0};