path: root/src/Startup.c

 /*@@
   @file      Startup.c
   @date      Mon Jun 19 2000
   @author    Thomas Radke
   @desc 
              Startup and termination routines for IOStreamedHDF5.
   @enddesc 
   @version   $Id$
 @@*/


#include <stdlib.h>

#include "cctk.h"
#include "cctk_Parameters.h"
#include "util_Network.h"
#include "BetaThorns/Socket/src/SocketUtils.h"
#include "CactusBase/IOUtil/src/ioutil_AdvertisedFiles.h"
#include "CactusBase/IOUtil/src/ioutil_CheckpointRecovery.h"
#include "ioStreamedHDF5GH.h"

#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif

/* the rcs ID and its dummy function to use it */
static char *rcsid = "$Id$";
CCTK_FILEVERSION(BetaThorns_IOStreamedHDF5_Startup_c)


/* prototypes of routines defined in this source file */
void IOStreamedHDF5_Startup (void);
void IOStreamedHDF5_Terminate (cGH *GH);
static void *IOStreamedHDF5_SetupGH (tFleshConfig *config,
                                     int convergence_level,
                                     cGH *GH);


 /*@@
   @routine   IOStreamedHDF5_Startup
   @date      Mon Jun 19 2000
   @author    Thomas Radke
   @desc 
              The startup registration routine for IOStreamedHDF5.
              Registers the GH extensions needed for IOStreamedHDF5
              along with its setup routine.
   @enddesc 

   @calls     CCTK_RegisterGHExtensionSetupGH
@@*/
void IOStreamedHDF5_Startup (void)
{
  /* check that thorn IOHDF5Util was activated */
  if (CCTK_GHExtensionHandle ("IOHDF5Util") < 0)
  {
    CCTK_WARN (1, "Thorn IOHDF5Util was not activated. "
                  "No IOStreamedHDF5 IO methods will be registered.");
    return;
  }

  /* FIXME: this check can go as soon as the Stream VFD
            comes with the default HDF5 installation */
#ifndef H5_HAVE_STREAM
  CCTK_WARN (1, "The Stream VFD was not configured in your HDF5 installation. "
                "No HDF5 streaming IO will be available !");
#else
  CCTK_RegisterGHExtensionSetupGH (CCTK_RegisterGHExtension ("IOStreamedHDF5"),
                                   IOStreamedHDF5_SetupGH);
#endif
}


 /*@@
   @routine   IOStreamedHDF5_Terminate
   @date      Mon Jun 19 2000
   @author    Thomas Radke
   @desc 
              IOStreamedHDF5's termination routine
              Closes all open sockets and destroys the timers.
   @enddesc 

   @calls     CCTK_TimerDestroyI

   @var       GH
   @vdesc     Pointer to CCTK grid hierarchy
   @vtype     cGH *
   @vio       in
   @endvar
@@*/
void IOStreamedHDF5_Terminate (cGH *GH)
{
  int i;
  ioStreamedHDF5GH *myGH;


  myGH = (ioStreamedHDF5GH *) CCTK_GHExtension (GH, "IOStreamedHDF5");
  if (myGH)
  {
    /* close the data and checkpoint output sockets */
    if (myGH->data_socket >= 0)
    {
      close (myGH->data_socket);
    }
    if (myGH->checkpoint_socket >= 0)
    {
      close (myGH->checkpoint_socket);
    }

    /* release allocated timers */
    if (myGH->print_timing_info)
    {
      for (i = 0; i < IOHDF5_NUM_TIMERS; i++)
      {
        CCTK_TimerDestroyI (myGH->timers[i]);
      }
    }

    /* remove advertised file and free filename */
    if (myGH->advertised_filename)
    {
      remove (myGH->advertised_filename);
      free (myGH->advertised_filename);
    }
  }
}


/****************************************************************************/
/*                           local routines                                 */
/****************************************************************************/
 /*@@
   @routine   IOStreamedHDF5_SetupGH
   @date      Mon Jun 19 2000
   @author    Thomas Radke
   @desc 
              Allocates and sets up IOStreamedHDF5's GH extension structure
   @enddesc 

   @calls     CCTK_RegisterIOMethod
              CCTK_RegisterIOMethodOutputGH
              CCTK_RegisterIOMethodOutputVarAs
              CCTK_RegisterIOMethodTimeToOutput
              CCTK_RegisterIOMethodTriggerOutput
              Socket_TCPOpenServerSock
              Socket_SetNonBlocking
              IOUtil_AdvertiseFile
              Util_GetHostName
              CCTK_TimerCreateI
              CCTK_TimerDestroyI
              CCTK_TimerResetI

   @var       config
   @vdesc     the CCTK configuration as provided by the flesh
   @vtype     tFleshConfig *
   @vio       unused
   @endvar
   @var       convergence_level
   @vdesc     the convergence level
   @vtype     int
   @vio       unused
   @endvar
   @var       GH
   @vdesc     Pointer to CCTK grid hierarchy
   @vtype     cGH *
   @vio       in
   @endvar

   @returntype  void *
   @returndesc
                pointer to the allocated GH extension structure
   @endreturndesc
@@*/
static void *IOStreamedHDF5_SetupGH (tFleshConfig *config,
                                     int convergence_level,
                                     cGH *GH)
{
  DECLARE_CCTK_PARAMETERS
  int i;
  int myproc;
  int numvars;
  char hostname[256];
  ioStreamedHDF5GH *myGH;
  FILE *advertised_file_fd;
  ioAdvertisedFileDesc advertised_file;


  /* suppress compiler warnings about unused variables */
  config = config;
  convergence_level = convergence_level;

  myproc = CCTK_MyProc (GH);

  /* Register IOStreamedHDF5 routines as output methods  */
  i = CCTK_RegisterIOMethod ("IOStreamedHDF5");
  CCTK_RegisterIOMethodOutputGH (i, IOStreamedHDF5_OutputGH);
  CCTK_RegisterIOMethodOutputVarAs (i, IOStreamedHDF5_OutputVarAs);
  CCTK_RegisterIOMethodTimeToOutput (i, IOStreamedHDF5_TimeFor);
  CCTK_RegisterIOMethodTriggerOutput (i, IOStreamedHDF5_TriggerOutput);

  /* Register the IOStreamedHDF5 recovery routine to thorn IOUtil */
  if (IOUtil_RegisterRecover ("IOStreamedHDF5 recovery",
                              IOStreamedHDF5_Recover) < 0)
  {
    CCTK_WARN (1, "Failed to register IOStreamedHDF5 recovery routine");
  }

  numvars = CCTK_NumVars ();
  myGH = (ioStreamedHDF5GH *) malloc (sizeof (ioStreamedHDF5GH));
  myGH->geo_output = (ioHDF5Geo_t **) calloc (numvars, sizeof (ioHDF5Geo_t *));
  myGH->out_last = (int *) malloc (numvars * sizeof (int));
  myGH->advertised_filename = NULL;

  for (i = 0; i < numvars; i++)
  {
    myGH->out_last[i] = -1;
  }

  /* only processor 0 is doing socket I/O */
  myGH->data_socket = -1;
  if (myproc == 0)
  {
    myGH->data_socket = Socket_TCPOpenServerSock (data_port);
    if (myGH->data_socket < 0)
    {
      CCTK_VWarn (1, __LINE__, __FILE__, CCTK_THORNSTRING,
                  "Couldn't open TCP server socket on output port %d. "
                  "No HDF5 streaming output will be available !", data_port);
    }
    else if (Socket_SetNonBlocking (myGH->data_socket) < 0)
    {
      CCTK_WARN (1, "Couldn't set output socket into non-blocking mode. "
                    "No HDF5 streaming output will be available !");
      close (myGH->data_socket);
      myGH->data_socket = -1;
    }
    else
    {
      Util_GetHostName (hostname, sizeof (hostname));
      CCTK_VInfo (CCTK_THORNSTRING,
                  "HDF5 data streaming service started on\n"
                  "                         %s:%d",
                  hostname, data_port);

      /* write the hostname/portnumber information in a temporary file
         which gets advertised and then can be downloaded */
      myGH->advertised_filename = (char *) malloc (L_tmpnam);
      if (myGH->advertised_filename && tmpnam (myGH->advertised_filename))
      {
        advertised_file_fd = fopen (myGH->advertised_filename, "w");

        if (advertised_file_fd)
        {
          fprintf (advertised_file_fd, "Hostname:  %s\n", hostname);
          fprintf (advertised_file_fd, "Data port: %d", data_port);
          fclose (advertised_file_fd);

          advertised_file.slice = "";
          advertised_file.thorn = CCTK_THORNSTRING;
          advertised_file.varname = "All variables";
          advertised_file.description = "Streamed HDF5 data";
          advertised_file.mimetype = "data/streamed-hdf5";

          IOUtil_AdvertiseFile (GH, myGH->advertised_filename,&advertised_file);
        }
        else
        {
          CCTK_WARN (2, "Couldn't create unique temporary file ! "
                        "HDF5 data streaming was not advertised.");
        }
      }
      else
      {
        CCTK_WARN (2, "Couldn't create unique temporary filename "
                      "for advertising the hostname/portnumber information !");
        free (myGH->advertised_filename);
        myGH->advertised_filename = NULL;
      }
    }
  }

  /* create timers if timing info was requested */
  myGH->print_timing_info = print_timing_info;
  if (myGH->print_timing_info)
  {
    for (i = 0; i < IOHDF5_NUM_TIMERS; i++)
    {
      if ((myGH->timers[i] = CCTK_TimerCreateI ()) < 0)
      {
        break;
      }
    }
    if (i != IOHDF5_NUM_TIMERS)
    {
      CCTK_WARN (1, "Could not create timers for checkpoint/recovery ! "
                    "No timer information will be available.");
      while (--i >= 0)
      {
        CCTK_TimerDestroyI (myGH->timers[i]);
      }
      myGH->print_timing_info = 0;
    }
    else
    {
      CCTK_TimerResetI (myGH->timers[CP_TOTAL_TIMER]);
      CCTK_TimerResetI (myGH->timers[RECOVERY_TIMER]);
    }
  }

  /* FIXME: this check can go as soon as the Stream VFD
            comes with the default HDF5 installation */
#ifdef H5_HAVE_STREAM
  /* only processor 0 is doing socket I/O */
  myGH->checkpoint_socket = -1;
  myGH->checkpoint_fapl = -1;
  if (myproc == 0)
  {
    myGH->checkpoint_socket = Socket_TCPOpenServerSock (checkpoint_port);
    if (myGH->checkpoint_socket < 0)
    {
      CCTK_VWarn (1, __LINE__, __FILE__, CCTK_THORNSTRING,
                  "Couldn't open TCP server socket on checkpoint port %d. "
                  "No IOStreamedHDF5 checkpointing will be available !",
                  checkpoint_port);
    }
    else if (Socket_SetNonBlocking (myGH->checkpoint_socket) < 0)
    {
      CCTK_WARN (1, "Couldn't set checkpoint socket into non-blocking mode. "
                    "No IOStreamedHDF5 checkpointing will be available !");
      close (myGH->checkpoint_socket);
      myGH->checkpoint_socket = -1;
    }
    else
    {
      H5FD_stream_fapl_t fapl;


      /* setup file access property list and select Stream VFD */
      fapl.increment = 0;
      fapl.socket = myGH->checkpoint_socket;
      fapl.do_socket_io = 1;
      fapl.backlog = 5;
      fapl.broadcast_fn  = NULL;
      fapl.broadcast_arg = NULL;
      IOHDF5_ERROR (myGH->checkpoint_fapl = H5Pcreate (H5P_FILE_ACCESS));
      IOHDF5_ERROR (H5Pset_fapl_stream (myGH->checkpoint_fapl, &fapl));
     
      Util_GetHostName (hostname, sizeof (hostname));
      CCTK_VInfo (CCTK_THORNSTRING,
                  "HDF5 checkpoint streaming service started on\n"
                  "                         %s:%d",
                  hostname, checkpoint_port);
    }
  }
#endif

  return (myGH);
}