coordinator.c File Reference

Ring consensus coordinator implementation. More...

Go to the source code of this file.

Data Structures
struct	consensus_coordinator
	Internal coordinator structure. More...

Macros
#define	CONSENSUS_ROUND_INTERVAL_NS   (5ULL * 60 * NS_PER_SEC_INT) /* 5 minutes */
#define	CONSENSUS_COLLECTION_DEADLINE_NS   (30ULL * NS_PER_SEC_INT) /* 30 seconds */

Typedefs
typedef struct consensus_coordinator	consensus_coordinator_t
	Internal coordinator structure.

Functions
asciichat_error_t	consensus_coordinator_create (const uint8_t my_id[16], const consensus_topology_t *topology, consensus_election_func_t election_func, void *election_context, consensus_coordinator_t **out_coordinator)
	Initialize coordinator with given parameters.
void	consensus_coordinator_destroy (consensus_coordinator_t *coordinator)
	Clean up coordinator resources.
asciichat_error_t	consensus_coordinator_process (consensus_coordinator_t *coordinator, uint32_t timeout_ms)
	Main orchestration loop.
asciichat_error_t	consensus_coordinator_on_ring_members (consensus_coordinator_t *coordinator, const consensus_topology_t *new_topology)
	Update ring topology when participants change.
asciichat_error_t	consensus_coordinator_on_collection_start (consensus_coordinator_t *coordinator, uint32_t round_id, uint64_t deadline_ns)
	Handle STATS_COLLECTION_START packet.
asciichat_error_t	consensus_coordinator_on_stats_update (consensus_coordinator_t *coordinator, const uint8_t sender_id[16], const participant_metrics_t *metrics, uint8_t num_metrics)
	Handle STATS_UPDATE packet.
asciichat_error_t	consensus_coordinator_on_election_result (consensus_coordinator_t *coordinator, const uint8_t host_id[16], const uint8_t backup_id[16])
	Handle ELECTION_RESULT packet.
asciichat_error_t	consensus_coordinator_get_current_host (const consensus_coordinator_t *coordinator, uint8_t out_host_id[16], uint8_t out_backup_id[16])
	Get currently elected host.
consensus_state_machine_t	consensus_coordinator_get_state (const consensus_coordinator_t *coordinator)
	Get current coordinator state.
uint64_t	consensus_coordinator_time_until_next_round (const consensus_coordinator_t *coordinator)
	Get time until next round.
int	consensus_coordinator_get_metrics_count (const consensus_coordinator_t *coordinator)
	Get count of metrics in current round.

Detailed Description

Ring consensus coordinator implementation.

Definition in file coordinator.c.

Macro Definition Documentation

CONSENSUS_COLLECTION_DEADLINE_NS

#define CONSENSUS_COLLECTION_DEADLINE_NS   (30ULL * NS_PER_SEC_INT) /* 30 seconds */

Definition at line 16 of file coordinator.c.

CONSENSUS_ROUND_INTERVAL_NS

#define CONSENSUS_ROUND_INTERVAL_NS   (5ULL * 60 * NS_PER_SEC_INT) /* 5 minutes */

Definition at line 15 of file coordinator.c.

Typedef Documentation

consensus_coordinator_t

typedef struct consensus_coordinator consensus_coordinator_t

Internal coordinator structure.

Function Documentation

consensus_coordinator_create()

asciichat_error_t consensus_coordinator_create	(	const uint8_t	my_id[16],
		const consensus_topology_t *	topology,
		consensus_election_func_t	election_func,
		void *	election_context,
		consensus_coordinator_t **	out_coordinator
	)

Initialize coordinator with given parameters.

Definition at line 42 of file coordinator.c.

                                                                                          {
  if (!my_id || !topology || !election_func || !out_coordinator) {
    return SET_ERRNO(
        ERROR_INVALID_PARAM, "Invalid parameter: my_id=%p, topology=%p, election_func=%p, out_coordinator=%p",
        (const void *)my_id, (const void *)topology, (const void *)election_func, (const void *)out_coordinator);
  }
 
  consensus_coordinator_t *coordinator = SAFE_MALLOC(sizeof(consensus_coordinator_t), consensus_coordinator_t *);
  if (!coordinator) {
    return SET_ERRNO(ERROR_MEMORY, "Failed to allocate coordinator");
  }
 
  memset(coordinator, 0, sizeof(*coordinator));
  memcpy(coordinator->my_id, my_id, 16);
  coordinator->topology = topology;
  coordinator->election_func = election_func;
  coordinator->election_context = election_context;
 
  /* Initialize state machine */
  asciichat_error_t err = consensus_state_create(my_id, topology, &coordinator->state);
  if (err != ASCIICHAT_OK) {
    SAFE_FREE(coordinator);
    return err;
  }
 
  /* Initialize round scheduling */
  coordinator->last_round_start_ns = time_get_ns();
  coordinator->next_round_id = 1;
  coordinator->has_stored_result = false;
 
  log_debug("Coordinator created for node %u, first round in 5 minutes", my_id[0]);
 
  *out_coordinator = coordinator;
  return ASCIICHAT_OK;
}

References consensus_state_create(), consensus_coordinator::election_context, consensus_coordinator::election_func, consensus_coordinator::has_stored_result, consensus_coordinator::last_round_start_ns, consensus_coordinator::my_id, consensus_coordinator::next_round_id, consensus_coordinator::state, time_get_ns(), and consensus_coordinator::topology.

Referenced by session_consensus_create().

consensus_coordinator_destroy()

void consensus_coordinator_destroy

(

consensus_coordinator_t *

coordinator

)

Clean up coordinator resources.

Definition at line 83 of file coordinator.c.

                                                                         {
  if (!coordinator) {
    return;
  }
  consensus_state_destroy(coordinator->state);
  SAFE_FREE(coordinator);
}

References consensus_state_destroy(), and consensus_coordinator::state.

Referenced by session_consensus_destroy().

consensus_coordinator_get_current_host()

asciichat_error_t consensus_coordinator_get_current_host	(	const consensus_coordinator_t *	coordinator,
		uint8_t	out_host_id[16],
		uint8_t	out_backup_id[16]
	)

Get currently elected host.

Definition at line 370 of file coordinator.c.

                                                                                                             {
  if (!coordinator || !out_host_id || !out_backup_id) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameter");
  }
 
  /* Try to get from current state first */
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
  if (current_state == CONSENSUS_STATE_ELECTION_COMPLETE) {
    asciichat_error_t err = consensus_state_get_elected_host(coordinator->state, out_host_id);
    if (err == ASCIICHAT_OK) {
      err = consensus_state_get_elected_backup(coordinator->state, out_backup_id);
    }
    if (err == ASCIICHAT_OK) {
      return ASCIICHAT_OK;
    }
  }
 
  /* Fall back to stored result */
  if (coordinator->has_stored_result) {
    memcpy(out_host_id, coordinator->stored_host_id, 16);
    memcpy(out_backup_id, coordinator->stored_backup_id, 16);
    return ASCIICHAT_OK;
  }
 
  return SET_ERRNO(ERROR_INVALID_STATE, "No election result available");
}

References consensus_state_get_current_state(), consensus_state_get_elected_backup(), consensus_state_get_elected_host(), consensus_coordinator::has_stored_result, consensus_coordinator::state, consensus_coordinator::stored_backup_id, and consensus_coordinator::stored_host_id.

Referenced by session_consensus_get_elected_host(), and session_consensus_is_ready().

consensus_coordinator_get_metrics_count()

int consensus_coordinator_get_metrics_count

(

const consensus_coordinator_t *

coordinator

)

Get count of metrics in current round.

Definition at line 429 of file coordinator.c.

                                                                                        {
  if (!coordinator || !coordinator->state) {
    return -1;
  }
  return consensus_state_get_metrics_count(coordinator->state);
}

References consensus_state_get_metrics_count(), and consensus_coordinator::state.

Referenced by session_consensus_get_metrics_count().

consensus_coordinator_get_state()

consensus_state_machine_t consensus_coordinator_get_state

(

const consensus_coordinator_t *

coordinator

)

Get current coordinator state.

Definition at line 401 of file coordinator.c.

                                                                                                      {
  if (!coordinator || !coordinator->state) {
    return CONSENSUS_STATE_FAILED;
  }
  return consensus_state_get_current_state(coordinator->state);
}

References consensus_state_get_current_state(), and consensus_coordinator::state.

Referenced by session_consensus_get_state().

consensus_coordinator_on_collection_start()

asciichat_error_t consensus_coordinator_on_collection_start	(	consensus_coordinator_t *	coordinator,
		uint32_t	round_id,
		uint64_t	deadline_ns
	)

Handle STATS_COLLECTION_START packet.

Definition at line 279 of file coordinator.c.

                                                                                  {
  if (!coordinator) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Coordinator is NULL");
  }
 
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
  if (current_state != CONSENSUS_STATE_IDLE) {
    return SET_ERRNO(ERROR_INVALID_STATE, "Cannot start collection, state is %d", current_state);
  }
 
  /* Start collection */
  asciichat_error_t err = consensus_state_start_collection(coordinator->state);
  if (err != ASCIICHAT_OK) {
    return err;
  }
 
  coordinator->next_round_id = round_id;
  coordinator->collection_deadline_ns = deadline_ns;
 
  /* Measure and add our metrics */
  err = measure_and_add_metrics(coordinator);
  if (err != ASCIICHAT_OK) {
    log_warn("Failed to measure metrics: %d", err);
  }
 
  log_dev("Collection started: round_id=%u, deadline in %u seconds", round_id,
          (unsigned int)((deadline_ns - time_get_ns()) / NS_PER_SEC_INT));
 
  return ASCIICHAT_OK;
}

References consensus_coordinator::collection_deadline_ns, consensus_state_get_current_state(), consensus_state_start_collection(), consensus_coordinator::next_round_id, consensus_coordinator::state, and time_get_ns().

Referenced by session_consensus_on_collection_start().

consensus_coordinator_on_election_result()

asciichat_error_t consensus_coordinator_on_election_result	(	consensus_coordinator_t *	coordinator,
		const uint8_t	host_id[16],
		const uint8_t	backup_id[16]
	)

Handle ELECTION_RESULT packet.

Definition at line 342 of file coordinator.c.

                                                                                                                   {
  if (!coordinator || !host_id || !backup_id) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameter");
  }
 
  /* Store the elected host and backup */
  memcpy(coordinator->stored_host_id, host_id, 16);
  memcpy(coordinator->stored_backup_id, backup_id, 16);
  coordinator->has_stored_result = true;
 
  log_info("Election result received: host=%u, backup=%u", host_id[0], backup_id[0]);
 
  /* Transition state back to IDLE */
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
  if (current_state == CONSENSUS_STATE_ELECTION_COMPLETE) {
    asciichat_error_t err = consensus_state_reset_to_idle(coordinator->state);
    if (err != ASCIICHAT_OK) {
      log_warn("Failed to reset state to IDLE: %d", err);
    }
  }
 
  return ASCIICHAT_OK;
}

References consensus_state_get_current_state(), consensus_state_reset_to_idle(), consensus_coordinator::has_stored_result, consensus_coordinator::state, consensus_coordinator::stored_backup_id, and consensus_coordinator::stored_host_id.

Referenced by session_consensus_on_election_result().

consensus_coordinator_on_ring_members()

asciichat_error_t consensus_coordinator_on_ring_members	(	consensus_coordinator_t *	coordinator,
		const consensus_topology_t *	new_topology
	)

Update ring topology when participants change.

Definition at line 250 of file coordinator.c.

                                                                                                  {
  if (!coordinator || !new_topology) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameter");
  }
 
  /* Update topology reference */
  coordinator->topology = new_topology;
 
  /* Reset state if we're in the middle of a round */
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
  if (current_state != CONSENSUS_STATE_IDLE) {
    log_warn("Ring topology changed during round, resetting state");
    consensus_state_destroy(coordinator->state);
    coordinator->state = NULL;
 
    asciichat_error_t err = consensus_state_create(coordinator->my_id, new_topology, &coordinator->state);
    if (err != ASCIICHAT_OK) {
      return err;
    }
  }
 
  log_info("Ring topology updated");
  return ASCIICHAT_OK;
}

References consensus_state_create(), consensus_state_destroy(), consensus_state_get_current_state(), consensus_coordinator::my_id, consensus_coordinator::state, and consensus_coordinator::topology.

Referenced by session_consensus_set_topology().

consensus_coordinator_on_stats_update()

asciichat_error_t consensus_coordinator_on_stats_update	(	consensus_coordinator_t *	coordinator,
		const uint8_t	sender_id[16],
		const participant_metrics_t *	metrics,
		uint8_t	num_metrics
	)

Handle STATS_UPDATE packet.

Definition at line 314 of file coordinator.c.

                                                                                                                   {
  if (!coordinator || !sender_id || !metrics) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameter");
  }
 
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
  if (current_state != CONSENSUS_STATE_COLLECTING) {
    return SET_ERRNO(ERROR_INVALID_STATE, "Cannot accept metrics, state is %d", current_state);
  }
 
  /* Add all metrics to state */
  for (int i = 0; i < num_metrics; i++) {
    asciichat_error_t err = consensus_state_add_metrics(coordinator->state, &metrics[i]);
    if (err != ASCIICHAT_OK) {
      log_warn("Failed to add metric %d: %d", i, err);
    }
  }
 
  log_debug("Received %d metrics from sender %u", num_metrics, sender_id[0]);
 
  return ASCIICHAT_OK;
}

References consensus_state_add_metrics(), consensus_state_get_current_state(), and consensus_coordinator::state.

Referenced by session_consensus_on_stats_update().

consensus_coordinator_process()

asciichat_error_t consensus_coordinator_process	(	consensus_coordinator_t *	coordinator,
		uint32_t	timeout_ms
	)

Main orchestration loop.

Definition at line 212 of file coordinator.c.

                                                                                                           {
  (void)timeout_ms; /* Currently unused, but kept for future use */
 
  if (!coordinator) {
    return SET_ERRNO(ERROR_INVALID_PARAM, "Coordinator is NULL");
  }
 
  uint64_t now_ns = time_get_ns();
 
  /* Check if we should start a new round */
  if (should_start_new_round(coordinator, now_ns)) {
    /* Only leader starts rounds */
    if (consensus_topology_am_leader(coordinator->topology)) {
      asciichat_error_t err = start_collection_round(coordinator, now_ns);
      if (err != ASCIICHAT_OK) {
        log_warn("Failed to start collection round: %d", err);
        return err;
      }
    }
  }
 
  consensus_state_machine_t current_state = consensus_state_get_current_state(coordinator->state);
 
  /* Check for collection timeout */
  if (current_state == CONSENSUS_STATE_COLLECTING && has_collection_timed_out(coordinator, now_ns)) {
    log_warn("Collection round timed out, completing early");
    asciichat_error_t err = complete_collection(coordinator);
    if (err != ASCIICHAT_OK) {
      log_error("Failed to complete collection on timeout: %d", err);
    }
  }
 
  return ASCIICHAT_OK;
}

References consensus_state_get_current_state(), consensus_topology_am_leader(), consensus_coordinator::state, time_get_ns(), and consensus_coordinator::topology.

Referenced by session_consensus_process().

consensus_coordinator_time_until_next_round()

uint64_t consensus_coordinator_time_until_next_round

(

const consensus_coordinator_t *

coordinator

)

Get time until next round.

Definition at line 411 of file coordinator.c.

                                                                                                 {
  if (!coordinator) {
    return 0;
  }
 
  uint64_t now_ns = time_get_ns();
  uint64_t next_round_ns = coordinator->last_round_start_ns + CONSENSUS_ROUND_INTERVAL_NS;
 
  if (now_ns >= next_round_ns) {
    return 0;
  }
 
  return next_round_ns - now_ns;
}

References CONSENSUS_ROUND_INTERVAL_NS, consensus_coordinator::last_round_start_ns, and time_get_ns().

Referenced by session_consensus_time_until_next_round().