libmsg/util/pingpong_ipc.c

#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <errno.h>
#include <time.h>
#include <signal.h>

#define DEFAULT_TIMER   60  /* Dur<75>e par d<>faut du test (en secondes) */
#define DEFAULT_WINSIZE 100   /* Taille par d<>faut de la fen<65>tre de messages */
#define DEFAULT_MSGSIZE 100   /* Taille par d<>faut de chaque message */

#define FATHER_KEY  64    /* Cl<43> du port de messages du p<>re */
#define CHILD_KEY   65    /* Cl<43> du port de messages du fils */

#ifndef FALSE
#define FALSE   0
#endif

#ifndef TRUE
#define TRUE  1
#endif

/*----------------------------------------------------------------*/

typedef struct {
  long 		mtype;
  char *	mtext;
} T_Msg;

int My_Port, Its_Port;
int Child_Pid, Father_Pid, Its_Pid, My_Pid;
int Verbose;
unsigned int Timer;
unsigned int Win_Size;
unsigned int Msg_Size;
int Cpt_Msg;
T_Msg ** Msg_Window;
int End_Test, End_Sigum;

char * error_code (void);
void stop_test (int);
void stop_father (void);
void stop_child (void);
void clean (int);
void trace (char *);
void print_result (void);
void parse_arg (int, char **);
extern char * strdup (const char *);

/*----------------------------------------------------------------*/
int main (int argc, char ** argv)
{
  unsigned int i, size;
  char str [256];
  T_Msg * Msg;

  /* R<>cup<75>ration des arguments de la ligne de commande */

  parse_arg (argc, argv);

  /* R<>cup<75>ration du no de process courant */

  Father_Pid = getpid ();

  /* Cr<43>ation d'un process fils avec qui dialoguer */

  if ((Child_Pid = fork ()) != 0)
  {
    /*--------- Code pour le process p<>re -----------*/

    Its_Pid = Child_Pid;
    My_Pid = Father_Pid;

    My_Port = 0;

    signal (SIGINT, stop_test);
    signal (SIGUSR1, stop_test);

    /* Cr<43>ation d'un port de messages */

    if ((My_Port = msgget (FATHER_KEY, 0777|IPC_CREAT|IPC_EXCL)) == -1)
    {
      fprintf (stderr, "Unable to create message queue for the father process : error %d (%s)\n", errno, error_code ());
      clean (TRUE);

      return 0;
    }
	else
    {
      sprintf (str, "Message port %d created", My_Port);
      trace (str);
    }

    /* On attend que le fils ait cr<63><72> son port de messages */

    sleep (2);

    /* Initialisation de la fen<65>tre de messages */

    Msg_Window = (T_Msg **)malloc (Win_Size);

    for (i = 0; i < Win_Size; i++)
    {
      Msg_Window[i] = (T_Msg *)malloc (sizeof (T_Msg));
      Msg_Window[i]->mtype = 1;
      Msg_Window[i]->mtext = (char *)malloc (Msg_Size);
      Msg_Window[i]->mtext[0] = (char)i;
    }

    /* R<>cup<75>ration de la port de messages du fils */

    if ((Its_Port = msgget (CHILD_KEY, 0)) == -1)
    {
      fprintf (stderr, "Unable to retrieve the message queue of the child process : error %d (%s)\n", errno, error_code ());
      clean (TRUE);

      return 0;
    }

    Msg = malloc (sizeof (T_Msg));
    Msg->mtext = (char *)malloc (Msg_Size);

    /* Initialisation du Timer */

    signal (SIGALRM, stop_test);

    alarm (Timer);

    /* D<>but du test : traitement des messages */

    fprintf (stdout, "Testing IPC message queue for %d second (s) with %d message (s) of %d byte (s)...\n", Timer, Win_Size, Msg_Size);

    /* Envoi de tous les messages de la fen<65>tre au fils */

    for (i = 0; i < Win_Size; i++)
    {
      errno = 0;

      if (msgsnd (Its_Port, Msg_Window[i], Msg_Size, IPC_NOWAIT) == 0)
      {
        if (Verbose)
        {
            sprintf (str, "Message %d sent to port %d", i, Its_Port);
            trace (str);
        }
      }

      if (errno != 0 && errno != EAGAIN)
        fprintf (stderr, "Unable to send message %d to port %d : error %d (%s)\n", i, Its_Port, errno, error_code ());
    }

    Cpt_Msg = 0;

    while (End_Test == FALSE)
    {
      /* R<>ception de messages */

      errno = 0;

      if ((size = msgrcv (My_Port, Msg, Msg_Size, 0, MSG_NOERROR)) > 0)
      {
        i = (int)(Msg->mtext[0]);
        if (Verbose)
        {
          sprintf (str, "Message %d received from port %d", i, My_Port);
          trace (str);
        }

        Cpt_Msg++;

        /* R<>envoi du message au fils */
        
        errno = 0;

        if (msgsnd (Its_Port, Msg_Window[i], Msg_Size, IPC_NOWAIT) == 0)
        {
          if (Verbose)
          {
            sprintf (str, "Message %d replied", i);
            trace (str);
          }
        }
    
        if (errno != 0 && errno != EAGAIN)
          fprintf (stderr, "Unable to reply message %d to port %d (%s)\n", i, Its_Port, error_code ());
      }
 
      if (errno != 0)
        fprintf (stderr, "Unable to receive a message from port %d (%s)\n", My_Port, error_code ());
    }

    stop_father ();
  }
  else
  {
    /*---------- Code pour le process fils ----------*/

    My_Pid = getpid ();
    Its_Pid = Father_Pid;
    My_Port = 0;

    /* Trap du signal SIGUSR1 envoy<6F> par le process p<>re <20> la fin du test */

    signal (SIGUSR1, stop_test);
    signal (SIGINT, stop_test);

    /* Cr<43>ation d'un port de messages */

    if ((My_Port = msgget (CHILD_KEY, 0777|IPC_CREAT|IPC_EXCL)) == -1)
    {
      fprintf (stderr, "Unable to create message queue for the child process (%s)\n", error_code ());
      clean (TRUE);

      return 0;
    }
	else
    {
      sprintf (str, "Message port %d created", My_Port);
      trace (str);
    }

    /* On attend que le p<>re ait cr<63><72> son port de messages */

    sleep (2);

    Msg_Window = (T_Msg **)malloc (Win_Size);

    for (i = 0; i < Win_Size; i++)
    {
      Msg_Window[i] = (T_Msg *)malloc (sizeof (T_Msg));
      Msg_Window[i]->mtype = 1;
      Msg_Window[i]->mtext = (char *)malloc (Msg_Size * sizeof (char));
      Msg_Window[i]->mtext[0] = (char)i;
    }

    /* R<>cup<75>ration du port de messages du p<>re */

    if ((Its_Port = msgget (FATHER_KEY, 0)) == -1)
    {
      fprintf (stderr, "Unable to retrieve the message queue of the father process (%s)\n", error_code ());
      clean (TRUE);

      return 0;
    }

    /* Traitement des messages */

    Msg = malloc (sizeof (T_Msg));
    Msg->mtext = (char *)malloc (Msg_Size * sizeof (char));

    while (End_Test == FALSE)
    {
      /* R<>ception de messages */

      errno = 0;

      if ((size = msgrcv (My_Port, Msg, Msg_Size, 0, MSG_NOERROR)) > 0)
      {
        i = (int)(Msg->mtext[0]);
        if (Verbose)
        {
          sprintf (str, "Message %d received from port %d", i, My_Port);
          trace (str);
        }

        Cpt_Msg++;

        /* R<>envoi du message au fils */
        
        errno = 0;

        if (msgsnd (Its_Port, Msg_Window[i], Msg_Size, IPC_NOWAIT) == 0)
        {
          if (Verbose)
          {
            sprintf (str, "Message %d replied", i);
            trace (str);
          }
        }
    
        if (errno != 0 && errno != EAGAIN)
          fprintf (stderr, "Unable to reply message %d to port %d (%s)\n", i, Its_Port, error_code ());
      }
 
      if (errno != 0)
        fprintf (stderr, "Unable to receive a message from port %d (%s)\n", My_Port, error_code ());
    }

    stop_child ();
  }

  return 0;
}

/*----------------------------------------------------------------*/
void stop_test (int signum)
{
  End_Sigum = signum;
  End_Test = TRUE;
}
/*----------------------------------------------------------------*/
void stop_father (void)
{
  switch (End_Sigum)
  {
    case SIGINT:

      /* Fin initi<74>e par l'utilisateur : m<>nage simple */

    case SIGUSR1:

      /* Fin anormale en provenance du fils : m<>nage simple */

      clean (FALSE);

      break;

    case SIGALRM:

      /* Fin normale : m<>nage + stop du fils + affichage du r<>ultat */

      clean (TRUE);
      print_result ();

      break;

    default:

      /* Fin anormale sur le process p<>re : m<>nage + avertit le process fils */

      clean (TRUE);

      break;
  }

  exit (1);
}

/*----------------------------------------------------------------*/
void print_result (void)
{
  /* Affichage du r<>sultat du test */

  fprintf (stdout, "%d message (s) exchanged (%d Msg/sec)\n", Cpt_Msg, (int)(Cpt_Msg / Timer));
}

/*----------------------------------------------------------------*/
void stop_child (void)
{
  switch (End_Sigum)
  {
    case SIGINT:

      /* Fin initi<74>e par l'utilisateur : m<>nage simple */

    case SIGUSR1:

      /* Fin initi<74>e par le process p<>re : m<>nage simple */

      clean (FALSE);

      break;

    default:

      /* Fin anormale sur le process fils : m<>nage + avertit le process p<>re */

      clean (TRUE);

      break;
  }

  exit (0);
}
/*----------------------------------------------------------------*/
void clean (int warn)
{
  unsigned int i;
  char str [256];

  /* On avertit l'autre process */

  if (warn == TRUE) kill (Its_Pid, SIGUSR1);

  if (Msg_Window != NULL)
  {
    for (i = 0; i < Win_Size; i++)
    {
      if (Msg_Window[i] != NULL)
      {
        if (Msg_Window[i]->mtext != NULL) free (Msg_Window[i]->mtext);
        free (Msg_Window[i]);
      }
    }
    free (Msg_Window);
  }

  /* Destruction de la port de messages */

  if (My_Port > 0)
  {
    if (Verbose)
    {
      sprintf (str, "Removing message queue %d...\n", My_Port);
      trace (str);
    }

    msgctl (My_Port, IPC_RMID, NULL);
  }
}

/*----------------------------------------------------------------*/
char * error_code (void)
{
  static char * str;

  if (str) free (str);

  switch (errno)
  {
    case EACCES:
      str = strdup ("EACCES");
      break;

    case EEXIST:
      str = strdup ("EEXIST");
      break;

    case EIDRM:
      str = strdup ("EIDRM");
      break;

    case ENOENT:
      str = strdup ("ENOENT");
      break;

    case ENOMEM:
      str = strdup ("ENOMEM");
      break;

    case ENOSPC:
      str = strdup ("ENOSPC");
      break;

    case EFAULT:
      str = strdup ("EFAULT");
      break;

    case EINTR:
      str = strdup ("EINTR");
      break;

    case EINVAL:
      str = strdup ("EINVAL");
      break;

    case E2BIG:
      str = strdup ("E2BIG");
      break;

    case ENOMSG:
      str = strdup ("ENOMSG");
      break;

    case EAGAIN:
      str = strdup ("EAGAIN");
      break;

    default:
      str = strdup ("unknown");
      break;
  }

  return str;
}

/*----------------------------------------------------------------*/
void trace (char *str)
{
  time_t dts;
  struct tm *dt;
  char current_date [30];
  char str_trace [256];

  /* Get current date */

  time (&dts);
  dt = localtime (&dts);
  sprintf (current_date, "%04d/%02d/%02d %02d:%02d:%02d", dt->tm_year + 1900, dt->tm_mon + 1, dt->tm_mday, dt->tm_hour, dt->tm_min, dt->tm_sec);

  sprintf (str_trace, "Process %d - %s : %s\n", My_Pid, current_date, str);
  fprintf (stderr, str_trace);
  fflush (stderr);
}

/*----------------------------------------------------------------*/
void parse_arg (int argc, char ** argv)
{
  int i;

  Verbose = FALSE;

  for (i = 1; i < argc; i++)
  {
    if (!strcmp (argv[i], "--help") || !strcmp (argv[i], "-h"))
    {
      fprintf (stdout, "Usage : %s [options]\n", argv[0]);
      fprintf (stdout,"options :\n");
      fprintf (stdout, "\t --help -h\n");
      fprintf (stdout, "\t --clean\n");
      fprintf (stdout, "\t --verbose\n");
      fprintf (stdout, "\t --duration <time in seconds> (default is %d)\n", DEFAULT_TIMER);
      fprintf (stdout, "\t --window <message number> (default is %d)\n", DEFAULT_WINSIZE);
      fprintf (stdout, "\t --message <message size> (default is %d bytes)\n", DEFAULT_MSGSIZE);
      exit (1);
    }

    if (!strcmp (argv[i], "--clean"))
    {
      if ((My_Port = msgget (FATHER_KEY, 0)) != -1) msgctl (My_Port, IPC_RMID, NULL);
      if ((Its_Port = msgget (CHILD_KEY, 0)) != -1) msgctl (Its_Port, IPC_RMID, NULL);
      exit (1);
    }

    if (!strcmp (argv[i], "--verbose"))
    {
      Verbose = TRUE;
      continue;
    }

    if (!strcmp (argv[i], "--duration"))
    {
      i++;
      if (i == argc)
      {
        fprintf (stderr, "Missing argument after %s\n", argv[i - 1]);
        exit (0);
      }
      Timer = atoi (argv[i]);
      continue;
    }

    if (!strcmp (argv[i], "--window"))
    {
      i++;
      if (i == argc)
      {
        fprintf (stderr, "Missing argument after %s\n", argv[i - 1]);
        exit (0);
      }
      Win_Size = atoi (argv[i]);
      continue;
    }

    if (!strcmp (argv[i], "--message"))
    {
      i++;
      if (i == argc)
      {
        fprintf (stderr, "Missing argument after %s\n", argv[i - 1]);
        exit (0);
      }
      Msg_Size = atoi (argv[i]);
      continue;
    }

    fprintf (stdout, "Unknown option '%s'\n", argv[i]);
    exit (0);
  }

  if (Timer <= 0) Timer = DEFAULT_TIMER;
  if (Win_Size <= 0) Win_Size = DEFAULT_WINSIZE;
  if (Msg_Size <= 0) Msg_Size = DEFAULT_MSGSIZE;
}