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
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	start_webrtc_client_threads (server_context_t *server_ctx, const char *client_id)
	Start threads for a WebRTC client after crypto initialization.
int	remove_client (server_context_t *server_ctx, const char *client_id)
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.
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.
bool	g_client_manager_rwlock_initialized

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 src/server/client.h.

Typedef Documentation

server_context_t

typedef struct server_context_t server_context_t

Definition at line 16 of file src/server/client.h.

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().

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_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.

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.

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 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

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

{0};

g_client_manager_rwlock_initialized

bool g_client_manager_rwlock_initialized

extern

Track whether g_client_manager_rwlock was successfully initialized.

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