rti_remote.h

Go to the documentation of this file.

#if defined STANDALONE_RTI
 
#ifndef RTI_REMOTE_H
#define RTI_REMOTE_H
 
#include "rti_common.h"
 
#ifdef __RTI_AUTH__
#include <openssl/rand.h> // For secure random number generation.
#include <openssl/hmac.h> // For HMAC authentication.
#endif
 
#include "lf_types.h"
#include "pqueue_tag.h"
#include "net_abstraction.h"

#define MAX_TIME_FOR_REPLY_TO_STOP_REQUEST SEC(30)


typedef struct federate_info_t {
  scheduling_node_t enclave;
  bool requested_stop;
  lf_thread_t thread_id;
  net_abstraction_t net;
  struct sockaddr_in UDP_addr;
  bool clock_synchronization_enabled;
  pqueue_tag_t* in_transit_message_tags;
  bool has_upstream_transient_federates;
  bool is_transient;
  tag_t effective_start_tag;
  tag_t max_intended_tag;
  bool neighbor_structure_received;
  int32_t number_of_downstream_transients;
  int32_t* downstream_transients;
} federate_info_t;

typedef enum clock_sync_stat { clock_sync_off, clock_sync_init, clock_sync_on } clock_sync_stat;

typedef enum federation_life_cycle_phase {
  startup_phase,   // Not all persistent federates have joined.
  execution_phase, // All persistent federates have joined.
  shutdown_phase   // Federation is shutting down.
} federation_life_cycle_phase;

typedef struct pqueue_delayed_grant_element_t {
  pqueue_tag_element_t base;
  uint16_t fed_id;     // Id of the federate with delayed grant of tag (in base)
  bool is_provisional; // Boolean recoding if the delayed grant is provisional
} pqueue_delayed_grant_element_t;

typedef pqueue_tag_t pqueue_delayed_grants_t;

typedef struct rti_remote_t {
  rti_common_t base;

  int64_t max_start_time;

  int64_t start_time_multiple;

  int num_feds_proposed_start;

  volatile bool all_federates_exited;

  volatile bool all_persistent_federates_exited;

  const char* federation_id;

  uint16_t user_specified_port;

  uint16_t final_port_UDP;

  int socket_descriptor_UDP;

  net_abstraction_t rti_net;

  lf_thread_t clock_thread;

  clock_sync_stat clock_sync_global_status;

  uint64_t clock_sync_period_ns;

  int32_t clock_sync_exchanges_per_interval;

  bool authentication_enabled;

  bool stop_in_progress;

  int32_t number_of_transient_federates;

  int32_t number_of_connected_transient_federates;

  federation_life_cycle_phase phase;

  pqueue_delayed_grants_t* delayed_grants;
} rti_remote_t;
 
extern int lf_critical_section_enter(environment_t* env);
 
extern int lf_critical_section_exit(environment_t* env);
 
/*
 * Indicator that one or more federates have reported an error on resigning.
 */
extern bool _lf_federate_reports_error;

void update_federate_next_event_tag_locked(uint16_t federate_id, tag_t next_event_tag);

void handle_port_absent_message(federate_info_t* sending_federate, unsigned char* buffer);

void handle_timed_message(federate_info_t* sending_federate, unsigned char* buffer);

void handle_latest_tag_confirmed(federate_info_t* fed);

void handle_next_event_tag(federate_info_t* fed);


void handle_stop_request_message(federate_info_t* fed);

void handle_stop_request_reply(federate_info_t* fed);


void handle_address_query(uint16_t fed_id);

void handle_address_ad(uint16_t federate_id);

void send_physical_clock(unsigned char message_type, federate_info_t* fed, socket_type_t socket_type);

void handle_physical_clock_sync_message(federate_info_t* my_fed, socket_type_t socket_type);

void* clock_synchronization_thread(void* noargs);

void* federate_info_thread_TCP(void* fed);

void send_reject(net_abstraction_t net_abs, rejection_code_t error_code);

void* respond_to_erroneous_connections(void* nothing);

void initialize_federate(federate_info_t* fed, uint16_t id);

void free_federate_info(federate_info_t* fed);

void reset_transient_federate(federate_info_t* fed);

int start_rti_server();

void wait_for_federates();

void usage(int argc, const char* argv[]);

int process_clock_sync_args(int argc, const char* argv[]);

int process_args(int argc, const char* argv[]);

void initialize_RTI(rti_remote_t* rti);
 
#endif // RTI_REMOTE_H
#endif // STANDALONE_RTI