owlps/infrastructure-centred/owlps-aggregator/owlps-aggregatord.c

/*
 * This file is part of the rtap localisation project.
 */


#include "owlps-aggregator.h"



char *program_name = NULL ;
cfg_t *cfg = NULL ; // Configuration structure
couple_list *couples = NULL ; // Computed data list



int main(int argc, char **argv)
{
  int ret = 0 ; // Program return value
  struct sigaction action ; // Signal handler structure
  pthread_t thread ; // Thread for aggregated data monitoring function
  unsigned int listening_port ;
  int sockfd ; // UDP listening socket

  program_name = argv[0] ;
  initialise_configuration(argc, argv) ;

  /* Set up signal handlers */
  sigemptyset(&action.sa_mask) ;
  action.sa_handler = sigint_handler ;
  sigaction(SIGINT, &action, NULL) ;
  action.sa_handler = sigterm_handler ;
  sigaction(SIGTERM, &action, NULL) ;

  /* Create UDP socket */
  listening_port = cfg_getint(cfg, "listening_port") ;
  if ((sockfd = create_udp_listening_socket(listening_port)) < 0)
    {
      fprintf(stderr,
              "Error! Cannot listen on port %u.\n", listening_port) ;
      return ERR_CREATING_SOCKET ;
    }

  /* Set up thread */
  pthread_create(&thread, NULL,
                 (void *) &monitor_couple_list, &couples) ;

  run = TRUE ;
  ret = read_loop(sockfd) ;

  (void) close(sockfd) ; // Close socket
  free_couple_list(&couples) ; // Clean list
  cfg_free(cfg) ; // Clean configuration

  printf("%s: end.\n", program_name) ;
  return ret ;
}



void initialise_configuration(int argc, char **argv)
{
  parse_config_file(argc, argv) ;
  parse_command_line(argc, argv) ;
  check_configuration() ;

#ifdef DEBUG
  /* Configuration printing */
  fprintf(stderr, "Configuration:\n") ;
  cfg_print(cfg, stderr) ;
#endif // DEBUG
}


void parse_config_file(int argc, char **argv)
{
  // Config file options for confuse
  cfg_opt_t opts[] =
    {
      CFG_INT("listening_port", AGGREGATE_DEFAULT_PORT, CFGF_NONE),

      // Port and IP address of the localisation server:
      CFG_INT("positioner_port", POSITIONER_DEFAULT_PORT, CFGF_NONE),
      CFG_STR("positioner_ip", POSITIONER_DEFAULT_IP, CFGF_NONE),

      CFG_STR("output_file", "", CFGF_NONE),

      // Timeouts (in milliseconds):
      CFG_INT("aggregate_timeout", DEFAULT_AGGREGATE_TIMEOUT, CFGF_NONE),
      CFG_INT("keep_timeout", DEFAULT_KEEP_TIMEOUT, CFGF_NONE),

      // Time between two list checks (in microseconds):
      CFG_INT("check_interval", DEFAULT_CHECK_INTERVAL, CFGF_NONE),

      CFG_END()
    } ;

  char *config_file ;

  // Option -f specifies a config file, so we search for it first
  int opt ;
  do
    opt = getopt(argc, argv, OPTIONS) ;
  while (opt != 'f' && opt != -1) ;
  if (opt == 'f')
    {
      config_file = malloc((strlen(optarg) + 1) * sizeof(char)) ;
      strcpy(config_file, optarg) ;
    }
  else // If -f isn't found, we use the default config file
    {
      config_file =
        malloc((strlen(DEFAULT_CONFIG_FILE) + 1) * sizeof(char)) ;
      strcpy(config_file, DEFAULT_CONFIG_FILE) ;
    }

  /* Parse config file */
  cfg = cfg_init(opts, CFGF_NONE) ; // Initialise options
  if (cfg_parse(cfg, config_file) == CFG_PARSE_ERROR)
    {
      free(config_file) ;
      exit(ERR_PARSING_CONFIG_FILE) ;
    }
  free(config_file) ;
}


void parse_command_line(int argc, char **argv)
{
  int opt ;

  optind = 1 ; // Rewind argument parsing

  while ((opt = getopt(argc, argv, OPTIONS)) != -1)
    {
      switch (opt)
	{
	case 'a' :
	  cfg_setint(cfg, "aggregate_timeout", strtol(optarg, NULL, 0)) ;
	  break ;
	case 'c' :
	  cfg_setint(cfg, "check_interval", strtol(optarg, NULL, 0)) ;
	  break ;
	case 'f' : // Config file
	  break ; // (already parsed)
        case 'h' :
          print_usage() ;
          exit(0) ;
	case 'i' :
	  cfg_setstr(cfg, "positioner_ip", optarg) ;
	  break ;
	case 'k' :
	  cfg_setint(cfg, "keep_timeout", strtol(optarg, NULL, 0)) ;
	  break ;
	case 'l' :
	  cfg_setint(cfg, "listening_port", strtol(optarg, NULL, 0)) ;
	  break ;
	case 'o' :
	  cfg_setstr(cfg, "output_file", optarg) ;
	  break ;
	case 'p' :
	  cfg_setint(cfg, "positioner_port", strtol(optarg, NULL, 0)) ;
	  break ;
	default :
	  print_usage() ;
	  exit(ERR_BAD_USAGE) ;
	}
    }
}


void check_configuration()
{
  // output_file //
  if (cfg_getstr(cfg, "output_file")[0] == '\0')
    {
      fprintf(stderr, "Error! You must specify an output file.\n") ;
      print_usage() ;
      exit(ERR_BAD_USAGE) ;
    }

  // positioner_ip //
  if (cfg_getstr(cfg, "positioner_ip")[0] == '\0')
    {
      fprintf(stderr, "Error! You must specify the IP address of the"
              " localisation server.\n") ;
      print_usage() ;
      exit(ERR_BAD_USAGE) ;
    }

  // aggregate_timeout //
  if (cfg_getint(cfg, "aggregate_timeout") < 0)
    {
#ifdef DEBUG
      fprintf(stderr, "Warning! aggregate_timeout cannot be negative:"
              " failing back to default value.\n") ;
#endif // DEBUG
      cfg_setint(cfg, "aggregate_timeout", DEFAULT_AGGREGATE_TIMEOUT) ;
    }

  // keep_timeout //
  if (cfg_getint(cfg, "keep_timeout") < 0)
    {
#ifdef DEBUG
      fprintf(stderr, "Warning! keep_timeout cannot be negative:"
              " failing back to default value.\n") ;
#endif // DEBUG
      cfg_setint(cfg, "keep_timeout", DEFAULT_KEEP_TIMEOUT) ;
    }

  // check_interval //
  if (cfg_getint(cfg, "check_interval") < 0)
    {
#ifdef DEBUG
      fprintf(stderr, "Warning! check_interval cannot be negative:"
              " failing back to default value.\n") ;
#endif // DEBUG
      cfg_setint(cfg, "check_interval", DEFAULT_CHECK_INTERVAL) ;
    }
}



/*
 * Reads packets while the program is not stopped.
 */
int read_loop(int sockfd)
{
  int ret = 0 ; // Return value
  int nread ; // recvfrom return value
  struct sockaddr_in client; // UDP client structure
  socklen_t client_len = sizeof(client) ; // Size of clients
  couple_message message ; // Message read on the socket
  char
    *ap_mac_string,     // Return pointers for mac_bytes_to_string()
    *mobile_mac_string,
    *mobile_ip_string ; // and ip_bytes_to_string()

  while (run)
    {
      nread = recvfrom(sockfd, &message, sizeof(message), 0,
                       (struct sockaddr *) &client, &client_len) ;

      if (nread <= 0)
	{
	  if (run)
	    {
	      fprintf(stderr, "No message received from client!\n") ;
	      ret = ERR_NO_MESSAGE_RECEIVED ;
	    }
	  break ;
	}
      ap_mac_string =
        mac_bytes_to_string(message.ap_mac_addr_bytes) ;
      mobile_mac_string =
        mac_bytes_to_string(message.mobile_mac_addr_bytes) ;
      mobile_ip_string =
        ip_bytes_to_string(message.mobile_ip_addr_bytes) ;
      printf("\n"
             "*** Message received from client ***\n"
             "\tAP MAC: %s\n"
             "\tMobile MAC: %s\n"
             "\tMobile IP: %s\n"
             "\tSequence number (request timestamp): %llu\n"
             "\tRequest arrival time on the AP: %llu\n"
             "\tSignal: %d dBm\n"
             "\tPosition X: %f\n"
             "\tPosition Y: %f\n"
             "\tPosition Z: %f\n"
             "\tDirection: %hhd\n"
             ,
	     ap_mac_string,
	     mobile_mac_string,
             mobile_ip_string,
	     timeval_to_ms(message.request_time),
	     timeval_to_ms(message.start_time),
	     message.antenna_signal_dbm - 0x100,
	     message.x_position,
	     message.y_position,
	     message.z_position,
	     message.direction
	     ) ;
      free(ap_mac_string) ;
      free(mobile_mac_string) ;
      free(mobile_ip_string) ;

      got_couple_info(&couples, message) ;
    }

  return ret ;
}



/*
 * Thread function. Monitors the list and sends information to the
 * localisation server when the timeout is reached.
 */
void* monitor_couple_list(couple_list **couples)
{
  couple_list *couple_ptr, *couple_prev ;
  couple_info_list *couple_info_ptr ;
  struct timeval current_time ;
  FILE *fd = NULL ;
  char *ap_mac_string ;
  unsigned long sub ; // sub_date() result

  unsigned long aggregate_timeout =
    (unsigned long) cfg_getint(cfg, "aggregate_timeout") ;
  unsigned long keep_timeout =
    (unsigned long) cfg_getint(cfg, "keep_timeout") ;

  struct sockaddr_in serv;
  struct sockaddr_in client ;
  socklen_t serv_len = sizeof(serv);
  request demande;
  couple_info info;
  int sockfd;

  sockfd = create_udp_sending_socket(cfg_getstr(cfg, "positioner_ip"),
                                     cfg_getint(cfg, "positioner_port"),
                                     &serv, &client) ;

  /* Open output file */
  fd = fopen(cfg_getstr(cfg, "output_file"), "a") ; // We use add mode
  if (fd == NULL) // If we failed to open the file,
    {
      perror("Cannot open output file") ;
      fprintf(stderr, "Redirecting output to standard output.") ;
      fd = stdout ; // we fail back to stdout.
    }

  while (run)
    {
      couple_ptr = *couples ;
      couple_prev = NULL ;
      couple_info_ptr = NULL ;
      gettimeofday(&current_time, NULL) ;

      while (couple_ptr != NULL) // Parsing list
	{
	  sub = sub_date(couple_ptr->start_time, current_time) ;

          // If the couple was not treated already
	  if (couple_ptr->info != NULL)
	    {
              // If the timeout is reached
	      if (sub > aggregate_timeout)
		{
		  printf("* Timeout reached.") ;
#ifdef DEBUG
		  printf(" sub=%lu > aggregate_timeout=%ld\n",
                         sub, aggregate_timeout) ;
#else // DEBUG
		  putchar('\n') ;
#endif // DEBUG

#ifdef TIMESTAMP
                  // Print request mobile timestamp to the output file
		  fprintf(fd, "%llu;",
                          timeval_to_ms(couple_ptr->request_time)) ;
#endif // TIMESTAMP
                  // Print couple info to the output file
		  fprintf(fd, "%0.2f;%0.2f;%0.2f;%hhd",
                          couple_ptr->x_position, couple_ptr->y_position,
                          couple_ptr->z_position, couple_ptr->direction) ;

		  memcpy(demande.mobile_mac_addr_bytes,
                         couple_ptr->mobile_mac_addr_bytes, 6) ;
		  demande.request_time = couple_ptr->request_time ;
		  demande.nb_couples = 0 ;

		  couple_info_ptr = couple_ptr->info ;
		  while (couple_info_ptr != NULL)
		    {
		      demande.nb_couples++;
		      couple_info_ptr = couple_info_ptr->next ;
                    }
		  sendto(sockfd, (void *)&demande, sizeof(request), 0,
                         (struct sockaddr *)&serv, serv_len) ;

                  // Send couples to the server and empty the list
		  couple_info_ptr = couple_ptr->info ;
		  while (couple_info_ptr != NULL)
		    {
		      // Send AP info to the localisation server
		      memcpy(info.ap_mac_addr_bytes,
                             couple_info_ptr->ap_mac_addr_bytes, 6) ;
		      info.antenna_signal_dbm =
                        couple_info_ptr->antenna_signal_dbm - 0x100 ;
		      sendto(sockfd, (void *)&info, sizeof(couple_info),
                             0, (struct sockaddr *)&serv, serv_len) ;

 		      // Print AP info to the output file
		      ap_mac_string =
                        mac_bytes_to_string(couple_info_ptr
                                            ->ap_mac_addr_bytes) ;
		      fprintf(fd, ";%s;%d", ap_mac_string,
                              couple_info_ptr->antenna_signal_dbm - 0x100) ;
		      free(ap_mac_string) ;

		      // Delete couple
		      couple_info_ptr = couple_info_ptr->next ;
		      free(couple_ptr->info) ;
		      couple_ptr->info = couple_info_ptr ;
		    }

		  fprintf(fd, "\n") ;
		}
	    }

          // If the couple was treated and keep timeout is reached
	  else if (sub > keep_timeout)
	    {
	      couple_list *couple_tmp = couple_ptr ;

	      printf("* Keep timeout reached.") ;
#ifdef DEBUG
	      printf(" sub=%lu > keep_timeout=%ld\n",
                     sub, keep_timeout) ;
#else // DEBUG
	      putchar('\n') ;
#endif // DEBUG

	      couple_ptr = couple_ptr->next ;

              // If it is the first couple of the list
	      if (couple_prev == NULL)
		*couples = couple_ptr ; // we shift the head
	      else // else we put the next of the previous on the next
		couple_prev->next = couple_ptr ;

	      free(couple_tmp) ;

	      continue ;
	    }

	  // Next couple
	  couple_prev = couple_ptr ;
	  couple_ptr = couple_ptr->next ;
	}

      fflush(NULL) ;
      usleep(cfg_getint(cfg, "check_interval")) ; // Wait to check again
    }

  /* Close output file */
  if (fclose(fd) != 0)
    perror("Error closing output file") ;

  return NULL ;
}



/*
 * Treats a received packet.
 */
void got_couple_info(couple_list **couples, couple_message message)
{
  couple_list *tmp_couple = NULL ;
  couple_info_list *tmp_info = NULL ;
  struct timeval start_time ; // Reception time on the aggregator

  gettimeofday(&start_time, NULL) ;

  /* Create a new couple */
  tmp_info = malloc(sizeof(couple_info_list)) ;
  memcpy(tmp_info->ap_mac_addr_bytes, message.ap_mac_addr_bytes, 6) ;
  tmp_info->antenna_signal_dbm = message.antenna_signal_dbm ;
  tmp_info->next = NULL ;

  /* Add it in the list */
  tmp_couple = *couples ;
  if (*couples == NULL) // If the couple list does not exist,
    {
      printf("Creating couple list.\n") ;
      tmp_couple = malloc(sizeof(couple_list)) ; // create it.
      memcpy(tmp_couple->mobile_mac_addr_bytes,
             message.mobile_mac_addr_bytes, 6) ;
      if (timeval_to_ms(message.request_time) != 0) // Explicit packet
        // Transmission time on the mobile:
        tmp_couple->request_time = message.request_time ;
      else // Implicit packet
        // Reception time on the AP:
        tmp_couple->request_time = message.start_time ;
      // Save locale time on the aggregator (not the reception time
      // on the AP):
      tmp_couple->start_time = start_time ;
      tmp_couple->x_position = message.x_position ;
      tmp_couple->y_position = message.y_position ;
      tmp_couple->z_position = message.z_position ;
      tmp_couple->direction = message.direction ;
      tmp_couple->next = NULL ;
      tmp_couple->info = tmp_info ;
      *couples = tmp_couple ;
    }

  else // If the couple list exists already
    { // we search the list for the couple
      if (timeval_to_ms(message.request_time) != 0) // Explicit packet
        {
          while (tmp_couple != NULL)
            { // Research criterion: MAC and transmission time
              if(mac_cmp(message.mobile_mac_addr_bytes,
                         tmp_couple->mobile_mac_addr_bytes) == 1
                 && sub_date(message.request_time,
                             tmp_couple->request_time) == 0)
                break ; // If the couple exists, we stop on it
              tmp_couple = tmp_couple->next ;
            }
        }
      else // Implicit packet
        {
          while (tmp_couple != NULL)
            { // Research criterion: MAC addresses equals and reception
              // times on the APs less than 10 ms
              // TODO : define an option for the maximal difference time.
              if(mac_cmp(message.mobile_mac_addr_bytes,
                         tmp_couple->mobile_mac_addr_bytes) == 1
                 && sub_date(message.start_time,
                             tmp_couple->request_time) <= 10)
                break ; // If the couple exists, we stop on it
              tmp_couple = tmp_couple->next ;
            }
        }

      if (tmp_couple == NULL) // If the couple does not exist in the list,
	{
	  printf("Create new couple.\n") ;
	  tmp_couple = malloc(sizeof(couple_list)) ; // create it.
	  memcpy(tmp_couple->mobile_mac_addr_bytes,
                 message.mobile_mac_addr_bytes, 6) ;
          if (timeval_to_ms(message.request_time) != 0) // Explicit packet
            // Transmission time on the mobile:
            tmp_couple->request_time = message.request_time ;
          else // Implicit packet
            // Reception time on the AP:
            tmp_couple->request_time = message.start_time ;
          // Save locale time on the aggregator (not the reception time
          // on the AP):
	  tmp_couple->start_time = start_time ;
	  tmp_couple->x_position = message.x_position ;
	  tmp_couple->y_position = message.y_position ;
	  tmp_couple->z_position = message.z_position ;
	  tmp_couple->direction = message.direction ;
	  tmp_couple->next = *couples ;
	  tmp_couple->info = tmp_info ;
	  *couples = tmp_couple ;
	}
      else // If the couple was found in the list
	{
	  if (tmp_couple->info == NULL)
	    { // We already sent to the server data for this couple
	      printf("Request already treated.\n") ;
	      free(tmp_info) ;
	    }
	  else
	    {
	      printf("Add information to the couple.\n") ;
	      tmp_info->next = tmp_couple->info ; // Add data
	      tmp_couple->info = tmp_info ;
	    }
	}
    }
}



/* Empty the couple list */
void free_couple_list(couple_list **couples)
{
  couple_list *couple_ptr = *couples ;
  couple_info_list *couple_info_ptr = NULL ;

  if (*couples != NULL)
    {
      while (couple_ptr != NULL)
	{
	  couple_info_ptr = couple_ptr->info ;
	  while (couple_info_ptr != NULL)
	    {
	      couple_info_ptr = couple_info_ptr->next ;
	      free(couple_ptr->info) ;
	      couple_ptr->info = couple_info_ptr ;
	    }
	  couple_ptr = couple_ptr->next ;
	  free(*couples) ;
	  *couples = couple_ptr ;
	}
    }
}



#ifdef DEBUG
/* Prints the couple list */
void print_couple_list(couple_list *couples)
{
  couple_list *couple_ptr = couples ;
  couple_info_list *info_ptr = NULL ;
  char *mobile_mac_string ;

  if (couples == NULL) // Empty list
    {
      printf("Aucun couple.\n") ;
      return ;
    }

  while (couple_ptr != NULL)
    {
      info_ptr = couple_ptr->info ; // Get the sub-list pointer

      mobile_mac_string =
        mac_bytes_to_string(couple_ptr->mobile_mac_addr_bytes) ;
      printf("Mobile MAC: %s\n"
             "Sequence number: %llu\n"
             "Reception timestamp: %llu\n"
             "\n",
	     mobile_mac_string,
	     timeval_to_ms(couple_ptr->request_time),
	     timeval_to_ms(couple_ptr->start_time)
	     ) ;
      free(mobile_mac_string) ;

      // Parse information relative to the current couple
      while (info_ptr != NULL)
	{
	  print_couple_info(info_ptr) ;
	  putchar('\n') ;
	  info_ptr = info_ptr->next ;
	}

      printf("\n\n") ;

      couple_ptr = couple_ptr->next ;
    }
}



/* Prints an element of a couple_info_list */
void print_couple_info(couple_info_list *info)
{
  char *ap_mac_string ;
  if (info == NULL)
    return ;

  ap_mac_string = mac_bytes_to_string(info->ap_mac_addr_bytes) ;
  printf("\tAP MAC: %s\n"
         "\tSignal strength: %d dBm\n",
	 ap_mac_string,
	 info->antenna_signal_dbm - 0x100
	 ) ;
  free(ap_mac_string) ;
}
#endif // DEBUG



char* ip_bytes_to_string(unsigned char *ip_binary)
{
  int
    taille = 16,
    i = 0 ;

  for (i = 0 ; i < 4 ; ++i)
    {
      if (ip_binary[i] < 0x64)
	taille-- ;
      if (ip_binary[i] < 0x10)
	taille-- ;
    }

  char *ret = malloc(sizeof(char) * taille) ;

  sprintf(ret, "%d.%d.%d.%d", ip_binary[0], ip_binary[1], ip_binary[2], ip_binary[3]) ;
  ret[taille-1] = '\0' ;

  return ret ;
}



void print_usage()
{
  printf("Usage:\n"
         "\t%s [-f config_file] [-l listening_port] [-i positionner_ip]"
         " [-p positioner_port] [-a aggregate_timeout] [-k keep_timeout]"
         " [-c check_interval] [-o output_file]\n"

         "Main options:\n"
         "\t-h\t\tPrint this help.\n"
         "\t-f config_file\tUse 'config_file' instead of the default"
         " configuration file.\n"

         "Server options:\n"
         "\t-l listening_port\tOnly requests sent on this port will be"
         " treated (default: %d).\n"
         "\t-i positionner_ip\tIP address of the localisation server"
         " (default: %s).\n"
         "\t-p positioner_port\tAggregated requests are transmitted to"
         " the localisation server on this port (default: %d).\n"

         "Aggregation options:\n"
         "\t-a aggregate_timeout\tRequests are stored during"
         " 'aggregate_timeout' before to be grouped (default:"
         " %d millisecondes).\n"
         "\t-k keep_timeout\t\tAggregated requests are kept during"
         " 'keep_timeout' milliseconds (default: %d milliseconds).\n"
         "\t-c check_interval\tTime between two checks of the stored"
         " requests (default\t%d microseconds).\n"
         ,
	 program_name,
	 AGGREGATE_DEFAULT_PORT,
	 POSITIONER_DEFAULT_IP,
	 POSITIONER_DEFAULT_PORT,
	 DEFAULT_AGGREGATE_TIMEOUT,
	 DEFAULT_KEEP_TIMEOUT,
	 DEFAULT_CHECK_INTERVAL
	 ) ;
}