server.h File Reference

🌐 Discovery server TCP connection manager More...

Go to the source code of this file.

Data Structures
struct	acds_client_data_t
	Per-client connection data. More...
struct	acds_server_t
	Discovery server state. More...

Functions
asciichat_error_t	acds_server_init (acds_server_t *server, const acds_config_t *config)
	Initialize discovery server.
asciichat_error_t	acds_server_run (acds_server_t *server)
	Run discovery server main loop.
void	acds_server_shutdown (acds_server_t *server)
	Shutdown discovery server.
void *	acds_client_handler (void *arg)
	Per-client connection handler (thread entry point)

Detailed Description

🌐 Discovery server TCP connection manager

Implements the core TCP server for the discovery service. Handles client connections, crypto handshakes, and dispatches packets to appropriate handlers (session management, WebRTC signaling).

Architecture

• Main thread accepts connections on TCP socket
• Per-client handler threads process ACIP packets
• Reuses lib/network/ for packet I/O
• Session registry protected by rwlock
• SQLite for persistence

Lifecycle

1. acds_server_init() - Create server, load identity, open database
2. acds_server_run() - Accept connections, spawn handlers
3. acds_server_shutdown() - Close sockets, stop threads, cleanup

Definition in file src/acds/server.h.

Function Documentation

acds_client_handler()

void * acds_client_handler

(

void *

arg

)

Per-client connection handler (thread entry point)

Processes ACIP packets from a connected client. Handles crypto handshake, then dispatches packets to session/signaling handlers.

Parameters

arg	Pointer to client connection context

Returns

NULL (thread exit value)

Definition at line 639 of file src/acds/server.c.

                                     {
  tcp_client_context_t *ctx = (tcp_client_context_t *)arg;
  if (!ctx) {
    log_error("Client handler: NULL context");
    return NULL;
  }
 
  acds_server_t *server = (acds_server_t *)ctx->user_data;
  socket_t client_socket = ctx->client_socket;
 
  // Get client IP for logging
  char client_ip[INET6_ADDRSTRLEN] = {0};
  tcp_client_context_get_ip(ctx, client_ip, sizeof(client_ip));
 
  /* Register this client handler thread with RCU library before any RCU operations.
     This ensures the thread is tracked in the RCU synchronization scheme for
     lock-free hash table access. */
  rcu_register_thread();
  log_info("Client handler started for %s (RCU registered)", client_ip);
 
  // Register client in TCP server registry with allocated client data
  acds_client_data_t *client_data = SAFE_MALLOC(sizeof(acds_client_data_t), acds_client_data_t *);
  if (!client_data) {
    tcp_server_reject_client(client_socket, "Failed to allocate client data");
    SAFE_FREE(ctx);
    rcu_unregister_thread();
    return NULL;
  }
  memset(client_data, 0, sizeof(*client_data));
  client_data->joined_session = false;
 
  if (tcp_server_add_client(&server->tcp_server, client_socket, client_data) != ASCIICHAT_OK) {
    SAFE_FREE(client_data);
    tcp_server_reject_client(client_socket, "Failed to register client in registry");
    SAFE_FREE(ctx);
    rcu_unregister_thread();
    return NULL;
  }
 
  log_debug("Client %s registered (socket=%d, total=%zu)", client_ip, client_socket,
            tcp_server_get_client_count(&server->tcp_server));
 
  // TODO: Crypto handshake (when crypto support is added)
  // For now, accept plain connections
 
  // Main packet processing loop
  while (atomic_load(&server->tcp_server.running)) {
    packet_type_t packet_type;
    void *payload = NULL;
    size_t payload_size = 0;
 
    // Receive packet (blocking with system timeout)
    int result = receive_packet(client_socket, &packet_type, &payload, &payload_size);
    if (result < 0) {
      // Connection error - client disconnected
      log_info("Client %s disconnected", client_ip);
      if (payload) {
        buffer_pool_free(NULL, payload, payload_size);
      }
      break;
    }
 
    log_debug("Received packet type 0x%02X from %s, length=%zu", packet_type, client_ip, payload_size);
 
    // O(1) ACIP array-based dispatch
    // Set server context for callbacks
    acip_acds_callbacks_t callbacks = g_acds_callbacks;
    callbacks.app_ctx = server;
 
    asciichat_error_t dispatch_result =
        acip_handle_acds_packet(NULL, packet_type, payload, payload_size, client_socket, client_ip, &callbacks);
 
    if (dispatch_result != ASCIICHAT_OK) {
      log_warn("ACIP handler failed for packet type 0x%02X from %s: %s", packet_type, client_ip,
               asciichat_error_string(dispatch_result));
    }
 
    // Free payload (allocated by receive_packet via buffer_pool_alloc)
    if (payload) {
      buffer_pool_free(NULL, payload, payload_size);
    }
  }
 
  // Cleanup
  tcp_server_remove_client(&server->tcp_server, client_socket);
  log_debug("Client %s unregistered (total=%zu)", client_ip, tcp_server_get_client_count(&server->tcp_server));
 
  socket_close(client_socket);
  SAFE_FREE(ctx);
 
  rcu_unregister_thread();
  log_info("Client handler finished for %s (RCU unregistered)", client_ip);
  return NULL;
}

References acip_handle_acds_packet(), acip_acds_callbacks_t::app_ctx, ASCIICHAT_OK, buffer_pool_free(), tcp_client_context_t::client_socket, acds_client_data_t::joined_session, log_debug, log_error, log_info, log_warn, receive_packet(), tcp_server::running, SAFE_FREE, SAFE_MALLOC, socket_close(), tcp_client_context_get_ip(), acds_server_t::tcp_server, tcp_server_add_client(), tcp_server_get_client_count(), tcp_server_reject_client(), tcp_server_remove_client(), and tcp_client_context_t::user_data.

Referenced by acds_server_init().

acds_server_init()

asciichat_error_t acds_server_init	(	acds_server_t *	server,
		const acds_config_t *	config
	)

Initialize discovery server.

Loads or generates identity keys, opens database, creates session registry, and binds TCP socket.

Parameters

server	Server structure to initialize
config	Configuration from command-line parsing

Returns

ASCIICHAT_OK on success, error code otherwise

Definition at line 86 of file src/acds/server.c.

                                                                                       {
  if (!server || !config) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "server or config is NULL");
  }
 
  memset(server, 0, sizeof(*server));
  memcpy(&server->config, config, sizeof(acds_config_t));
 
  // Initialize session registry
  server->sessions = SAFE_MALLOC(sizeof(session_registry_t), session_registry_t *);
  if (!server->sessions) {
    return SET_ERRNO(ERROR_MEMORY, "Failed to allocate session registry");
  }
 
  asciichat_error_t result = session_registry_init(server->sessions);
  if (result != ASCIICHAT_OK) {
    SAFE_FREE(server->sessions);
    return result;
  }
 
  // Open database
  result = database_init(config->database_path, &server->db);
  if (result != ASCIICHAT_OK) {
    session_registry_destroy(server->sessions);
    SAFE_FREE(server->sessions);
    return result;
  }
 
  // Load sessions from database
  result = database_load_sessions(server->db, server->sessions);
  if (result != ASCIICHAT_OK) {
    log_warn("Failed to load sessions from database (continuing anyway)");
  }
 
  // Initialize rate limiter with SQLite backend
  server->rate_limiter = rate_limiter_create_sqlite(NULL); // NULL = externally managed DB
  if (!server->rate_limiter) {
    database_close(server->db);
    session_registry_destroy(server->sessions);
    SAFE_FREE(server->sessions);
    return SET_ERRNO(ERROR_MEMORY, "Failed to create rate limiter");
  }
 
  // Set the database handle for the rate limiter
  rate_limiter_set_sqlite_db(server->rate_limiter, server->db);
 
  // Configure TCP server
  tcp_server_config_t tcp_config = {
      .port = config->port,
      .ipv4_address = (config->address[0] != '\0') ? config->address : NULL,
      .ipv6_address = (config->address6[0] != '\0') ? config->address6 : NULL,
      .bind_ipv4 = (config->address[0] != '\0') || (config->address[0] == '\0' && config->address6[0] == '\0'),
      .bind_ipv6 = (config->address6[0] != '\0') || (config->address[0] == '\0' && config->address6[0] == '\0'),
      .accept_timeout_sec = 1,
      .client_handler = acds_client_handler,
      .user_data = server,
  };
 
  // Initialize TCP server
  result = tcp_server_init(&server->tcp_server, &tcp_config);
  if (result != ASCIICHAT_OK) {
    rate_limiter_destroy(server->rate_limiter);
    database_close(server->db);
    session_registry_destroy(server->sessions);
    SAFE_FREE(server->sessions);
    return result;
  }
 
  // Initialize background worker thread pool
  atomic_store(&server->shutdown, false);
  server->worker_pool = thread_pool_create("acds_workers");
  if (!server->worker_pool) {
    log_warn("Failed to create worker thread pool");
    tcp_server_shutdown(&server->tcp_server);
    rate_limiter_destroy(server->rate_limiter);
    database_close(server->db);
    session_registry_destroy(server->sessions);
    SAFE_FREE(server->sessions);
    return SET_ERRNO(ERROR_MEMORY, "Failed to create worker thread pool");
  }
 
  // Spawn rate limit cleanup thread in worker pool
  if (thread_pool_spawn(server->worker_pool, cleanup_thread_func, server, 0, "rate_limit_cleanup") != ASCIICHAT_OK) {
    log_warn("Failed to spawn rate limit cleanup thread (continuing without cleanup)");
  }
 
  log_info("Discovery server initialized successfully");
  return ASCIICHAT_OK;
}

References acds_client_handler(), acds_config_t::address, acds_config_t::address6, ASCIICHAT_OK, acds_server_t::config, database_close(), database_init(), database_load_sessions(), acds_config_t::database_path, acds_server_t::db, ERROR_INVALID_PARAM, ERROR_MEMORY, log_info, log_warn, tcp_server_config_t::port, acds_config_t::port, acds_server_t::rate_limiter, rate_limiter_create_sqlite(), rate_limiter_destroy(), rate_limiter_set_sqlite_db(), SAFE_FREE, SAFE_MALLOC, session_registry_destroy(), session_registry_init(), acds_server_t::sessions, SET_ERRNO, acds_server_t::shutdown, acds_server_t::tcp_server, tcp_server_init(), tcp_server_shutdown(), thread_pool_create(), thread_pool_spawn(), and acds_server_t::worker_pool.

Referenced by main().

acds_server_run()

asciichat_error_t acds_server_run

(

acds_server_t *

server

)

Run discovery server main loop.

Accepts client connections and spawns handler threads. Blocks until shutdown signal received.

Parameters

server

Initialized server structure

Returns

ASCIICHAT_OK on success, error code otherwise

Definition at line 176 of file src/acds/server.c.

                                                         {
  if (!server) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "server is NULL");
  }
 
  log_info("Discovery server accepting connections on port %d", server->config.port);
 
  // Delegate to TCP server abstraction
  return tcp_server_run(&server->tcp_server);
}

References acds_server_t::config, ERROR_INVALID_PARAM, log_info, acds_config_t::port, SET_ERRNO, acds_server_t::tcp_server, and tcp_server_run().

Referenced by main().

acds_server_shutdown()

void acds_server_shutdown

(

acds_server_t *

server

)

Shutdown discovery server.

Closes listen socket, stops accepting connections, waits for handler threads to exit, closes database, and frees resources.

Parameters

server

Server structure to clean up

Definition at line 187 of file src/acds/server.c.

                                                 {
  if (!server) {
    return;
  }
 
  // Signal shutdown to worker threads
  atomic_store(&server->shutdown, true);
 
  // Shutdown TCP server (closes listen sockets, stops accept loop)
  tcp_server_shutdown(&server->tcp_server);
 
  // Wait for all client handler threads to exit
  // We need to stop all active client connections gracefully
  size_t remaining_clients;
  int shutdown_attempts = 0;
  const int max_shutdown_attempts = 100; // 10 seconds (100 * 100ms)
 
  while ((remaining_clients = tcp_server_get_client_count(&server->tcp_server)) > 0 &&
         shutdown_attempts < max_shutdown_attempts) {
    log_debug("Waiting for %zu client handler threads to exit (attempt %d/%d)", remaining_clients,
              shutdown_attempts + 1, max_shutdown_attempts);
    platform_sleep_ms(100);
    shutdown_attempts++;
  }
 
  if (remaining_clients > 0) {
    log_warn("Server shutdown: %zu client handler threads still running after 10 seconds", remaining_clients);
  } else if (shutdown_attempts > 0) {
    log_debug("All client handler threads exited gracefully");
  }
 
  // Stop and destroy worker thread pool (cleanup thread, etc.)
  if (server->worker_pool) {
    thread_pool_destroy(server->worker_pool);
    server->worker_pool = NULL;
    log_debug("Worker thread pool stopped");
  }
 
  // Destroy rate limiter
  if (server->rate_limiter) {
    rate_limiter_destroy(server->rate_limiter);
    server->rate_limiter = NULL;
  }
 
  // Close database
  if (server->db) {
    database_close(server->db);
    server->db = NULL;
  }
 
  // Destroy session registry
  if (server->sessions) {
    session_registry_destroy(server->sessions);
    SAFE_FREE(server->sessions);
  }
 
  log_info("Server shutdown complete");
}

References database_close(), acds_server_t::db, log_debug, log_info, log_warn, platform_sleep_ms(), acds_server_t::rate_limiter, rate_limiter_destroy(), SAFE_FREE, session_registry_destroy(), acds_server_t::sessions, acds_server_t::shutdown, acds_server_t::tcp_server, tcp_server_get_client_count(), tcp_server_shutdown(), thread_pool_destroy(), and acds_server_t::worker_pool.

Referenced by main().