---

nml_srv.cc

Go to the documentation of this file.

/*************************************************************************
* File:nml_srv.cc                                                        *
* Authors: Fred Proctor, Will Shackleford                                *
* Purpose: C++  file for RPC server that reads and writes                *
*          to a local NML buffer for remote processes.                   *
* Includes:                                                              *
*          1. class NML_SERVER member functions.                         *
*************************************************************************/

#include "rcs_defs.hh"          /* EXTERN_C_STD_HEADERS  */

#ifdef EXTERN_C_STD_HEADERS
extern "C"
{
#endif

#include <string.h>             /* memcpy() */

#if defined(WIN32) && !defined(gnuwin32)
#if defined(WIN32) && !defined(USE_OLD_WINSOCK)
// Lame problem if windows.h is included before winsock2.h many redefined
// compiler errors result.
#include <winsock2.h>
#endif

#include <windows.h>            /* GetCurrentProcessId(), GetCurrentThreadID() */
#else
#include <signal.h>             /* kill() */
#include <sys/types.h>
#include <unistd.h>             /* getpid() */
#include <sys/wait.h>           /* waitpid() */
#include <stdlib.h>             /* atexit() */
#ifdef VXWORKS
#include <taskLib.h>            /* taskIdSelf() */
#endif
#endif

#ifdef EXTERN_C_STD_HEADERS
}
#endif

#include "nml.hh"
#include "nmlmsg.hh"
#include "cms.hh"
#include "nml_srv.hh"
#include "rem_msg.hh"           /* struct REMOTE_READ_REQUEST, */
                                /* struct REMOTE_WRITE_REQUEST, */
                                /* struct REMOTE_READ_REPLY, */
                                /* struct REMOTE_WRITE_REPLY, */
#include "rcs_prnt.hh"          /* rcs_print_error() */
#include "timer.hh"             // esleep()
#include "rcs_exit.hh"          // rcs_exit
#include "dbg_mem.h"            // DEBUG_MALLOC, DEBUG_FREE
#include "cmsdiag.hh"

#ifdef sunos5
#include <thread.h>             // thr_exit
#endif


NML_SERVER::NML_SERVER (NML * _nml, int _set_to_master):
CMS_SERVER ()
{
  NML_SERVER_LOCAL_PORT *new_local_port = NULL;
  being_deleted = 0;
  if (NULL != _nml)
    {
      if (NULL != _nml->cms)
        {
          if (CMS_REMOTE_TYPE != _nml->cms->ProcessType)
            {
              NML *new_nml;
              if (_nml->cms->isserver &&
                  (0 == _set_to_master ||
                   (_nml->cms->is_local_master == 1 && _set_to_master == 1) ||
                   (_nml->cms->is_local_master == 0 && _set_to_master == -1)))
                {
                  new_nml = _nml;
                  if (NULL != new_nml)
                    {
                      new_local_port = new NML_SERVER_LOCAL_PORT (new_nml);
                      add_local_port (new_local_port);
                    }
                  new_local_port->local_channel_reused = 1;
                }
              else
                {
                  new_nml = new NML (_nml, 1, -1);
                  if (NULL != new_nml)
                    {
                      new_local_port = new NML_SERVER_LOCAL_PORT (new_nml);
                      add_local_port (new_local_port);
                    }
                  new_local_port->local_channel_reused = 0;
                }
            }
          else
            {
              rcs_print_error
                ("NML_SERVER:(ERROR) ProcessType was REMOTE.\n");
              _nml = (NML *) NULL;
            }
        }
      else
        {
          rcs_print_error ("NML_SERVER:(ERROR) cms was NULL.\n");
        }
    }
  else
    {
      rcs_print_error ("NML_SERVER:(ERROR) nml_ptr was NULL.\n");
    }
  add_to_nml_server_list ();
}

void
NML_SERVER::add_to_nml_server_list ()
{
  if (NULL == NML_Default_Super_Server)
    {
      NML_Default_Super_Server = new NML_SUPER_SERVER;
    }
  if (NULL != NML_Default_Super_Server)
    {
      NML_Default_Super_Server->add_to_list (this);
    }
}

NML_SERVER::~NML_SERVER ()
{
  being_deleted = 1;
  delete_from_list ();
}

void
NML_SERVER::delete_from_list ()
{
  CMS_SERVER::delete_from_list ();
  if (NULL != NML_Default_Super_Server)
    {
      if (NULL != NML_Default_Super_Server->servers)
        {
          NML_Default_Super_Server->servers->
            delete_node (super_server_list_id);
        }
    }
}

NML_SERVER_LOCAL_PORT::NML_SERVER_LOCAL_PORT (NML * _nml):
CMS_SERVER_LOCAL_PORT ((CMS *) NULL)
{
  local_channel_reused = 1;
  nml = _nml;
  if (NULL != nml)
    {
      cms = nml->cms;
      if (NULL != cms)
        {
          buffer_number = cms->buffer_number;
        }
    }
}

NML_SERVER_LOCAL_PORT::~NML_SERVER_LOCAL_PORT ()
{
  if (NULL != nml && !local_channel_reused)
    {
      delete nml;
    }
  nml = (NML *) NULL;
  cms = (CMS *) NULL;
}

REMOTE_READ_REPLY *
NML_SERVER_LOCAL_PORT::reader (REMOTE_READ_REQUEST * _req)
{
  if ((NULL == cms) || (NULL == nml))
    {
      rcs_print_error ("NMLserver:reader: CMS object is NULL.\n");
      return ((REMOTE_READ_REPLY *) NULL);
    }

  /* Setup CMS channel from request arguments. */
  cms->in_buffer_id = _req->last_id_read;

  /* Read and encode the buffer. */
  switch (_req->access_type)
    {
    case CMS_READ_ACCESS:
      nml->read ();
      break;
    case CMS_PEEK_ACCESS:
      nml->peek ();
      break;
    default:
      rcs_print_error ("NML_SERVER: Invalid access type.(%d)\n",
                       _req->access_type);
      break;
    }


  /* Setup reply structure to be returned to remote process. */
  read_reply.status = (int) cms->status;
  if (cms->status == CMS_READ_OLD)
    {
      read_reply.size = 0;
      read_reply.data = NULL;
      read_reply.write_id = _req->last_id_read;
      read_reply.was_read = 1;
    }
  else
    {
      read_reply.size = cms->header.in_buffer_size;
      read_reply.data = (unsigned char *) cms->encoded_data;
      read_reply.write_id = cms->in_buffer_id;
      read_reply.was_read = cms->header.was_read;
    }

  /* Reply structure contains the latest shared memory info-- now
     return it to cms_dispatch  for return to caller */
  return (&read_reply);
}


REMOTE_READ_REPLY *
NML_SERVER_LOCAL_PORT::blocking_read (REMOTE_READ_REQUEST * _req)
{
  if ((NULL == cms) || (NULL == nml))
    {
      rcs_print_error ("NMLserver:blocking_read: CMS object is NULL.\n");
      return ((REMOTE_READ_REPLY *) NULL);
    }
  nml->cms->first_diag_store = 0;
  if (_req->type != REMOTE_CMS_BLOCKING_READ_REQUEST_TYPE)
    {
      rcs_print_error ("NMLserver::blocking_read: Invalid request type(%d)\n",
                       _req->type);
      return NULL;
    }
  double orig_bytes_moved = 0.0;

  REMOTE_BLOCKING_READ_REQUEST *breq = (REMOTE_BLOCKING_READ_REQUEST *) _req;
  breq->_nml = new NML (nml, 1, -1);
  NML *nmlcopy = (NML *) breq->_nml;
  CMS *cmscopy = nmlcopy->cms;
  double blocking_timeout = (double) (breq->timeout_millis / 1000.0);
  REMOTE_READ_REPLY *temp_read_reply = new REMOTE_READ_REPLY ();
  breq->_reply = temp_read_reply;
  long data_size = (long) cmscopy->max_encoded_message_size;
  temp_read_reply->data = DEBUG_MALLOC (data_size);
  breq->_data = temp_read_reply->data;
  if ((NULL == cmscopy) || (NULL == nmlcopy))
    {
      rcs_print_error ("NMLserver:blocking_read: CMS object is NULL.\n");
      return ((REMOTE_READ_REPLY *) NULL);
    }
  if (NULL != cmscopy->handle_to_global_data)
    {
      orig_bytes_moved = cmscopy->handle_to_global_data->total_bytes_moved;
    }
  if (NULL == temp_read_reply->data)
    {
      rcs_print_error
        ("NMLserver:blocking_read: temp_read_reply->data object is NULL.\n");
      return ((REMOTE_READ_REPLY *) NULL);
    }
  nmlcopy->cms->set_encoded_data (temp_read_reply->data, data_size);

  /* Setup CMS channel from request arguments. */
  cmscopy->in_buffer_id = _req->last_id_read;

  /* Read and encode the buffer. */
  nmlcopy->blocking_read (blocking_timeout);

  /* Setup reply structure to be returned to remote process. */
  temp_read_reply->status = (int) cmscopy->status;
  if (cmscopy->status == CMS_READ_OLD)
    {
      temp_read_reply->size = 0;
      if (NULL != temp_read_reply->data)
        {
          breq->_data = NULL;
          DEBUG_FREE (temp_read_reply->data);
          temp_read_reply->data = NULL;
        }
      temp_read_reply->write_id = _req->last_id_read;
      temp_read_reply->was_read = 1;
    }
  else
    {
      temp_read_reply->size = cmscopy->header.in_buffer_size;
      temp_read_reply->write_id = cmscopy->in_buffer_id;
      temp_read_reply->was_read = cmscopy->header.was_read;
    }
  if (NULL != nml->cms->handle_to_global_data &&
      NULL != cmscopy->handle_to_global_data)
    {
      nml->cms->handle_to_global_data->total_bytes_moved
        +=
        (cmscopy->handle_to_global_data->total_bytes_moved -
         orig_bytes_moved);
      nml->cms->first_diag_store = cmscopy->first_diag_store;
    }
  breq->_nml = NULL;
  delete nmlcopy;


  /* Reply structure contains the latest shared memory info-- now
     return it to cms_dispatch  for return to caller */
  return (temp_read_reply);
}


/* REMOTE local_port function for writes */
REMOTE_WRITE_REPLY *
NML_SERVER_LOCAL_PORT::writer (REMOTE_WRITE_REQUEST * _req)
{
  NMLmsg *temp;                 /* Temporary Pointer */

  if ((NULL == cms) || (NULL == nml))
    {
      rcs_print_error ("NMLserver:writer: CMS object is NULL.\n");
      return ((REMOTE_WRITE_REPLY *) NULL);
    }

  temp = (NMLmsg *) cms->data;
  /* Check to see if remote process writing too much into local buffer. */
  if (_req->size > cms_encoded_data_explosion_factor * cms->size)
    {
      rcs_print_error
        ("CMSserver:cms_writer: CMS buffer size is too small.\n");
      return ((REMOTE_WRITE_REPLY *) NULL);
    }

  /* Copy the encoded data to the location set up in CMS. */
  // memcpy(cms->encoded_data, _req->data, _req->size);
  cms->header.in_buffer_size = _req->size;
  temp->size = _req->size;

  switch (_req->access_type)
    {
    case CMS_WRITE_ACCESS:
      nml->write (*temp);
      break;
    case CMS_WRITE_IF_READ_ACCESS:
      nml->write_if_read (*temp);
      break;
    default:
      rcs_print_error ("NML_SERVER: Invalid Access type. (%d)\n",
                       _req->access_type);
      break;
    }

  write_reply.status = (int) cms->status;
  write_reply.was_read = cms->header.was_read;
  return (&write_reply);
}



REMOTE_SET_DIAG_INFO_REPLY *
NML_SERVER_LOCAL_PORT::set_diag_info (REMOTE_SET_DIAG_INFO_REQUEST * _req)
{
  if (NULL == _req)
    {
      return (NULL);
    }
  CMS_DIAG_PROC_INFO *dpi = cms->get_diag_proc_info ();
  if (NULL == dpi)
    {
      return (NULL);
    }
  if (orig_info == NULL)
    {
      orig_info = new CMS_DIAG_PROC_INFO ();
      *orig_info = *dpi;
    }
  strncpy (dpi->name, _req->process_name, 16);
  strncpy (dpi->host_sysinfo, _req->host_sysinfo, 32);
  if (cms->total_connections > _req->c_num && _req->c_num >= 0)
    {
      cms->connection_number = _req->c_num;
    }
  if (NULL != cms->handle_to_global_data)
    {
      cms->handle_to_global_data->total_bytes_moved = _req->bytes_moved;
    }
  dpi->pid = _req->pid;
  dpi->rcslib_ver = _req->rcslib_ver;
  cms->set_diag_proc_info (dpi);
  return (NULL);
}

REMOTE_GET_DIAG_INFO_REPLY *
NML_SERVER_LOCAL_PORT::get_diag_info (REMOTE_GET_DIAG_INFO_REQUEST * _req)
{
  get_diag_info_reply.cdi = cms->get_diagnostics_info ();
  get_diag_info_reply.status = cms->status;
  return (&get_diag_info_reply);
}

REMOTE_GET_MSG_COUNT_REPLY *
NML_SERVER_LOCAL_PORT::get_msg_count (REMOTE_GET_DIAG_INFO_REQUEST * _req)
{
  return (NULL);
}

void
NML_SERVER_LOCAL_PORT::reset_diag_info ()
{
  if (NULL != orig_info)
    {
      CMS_DIAG_PROC_INFO *dpi = cms->get_diag_proc_info ();
      *dpi = *orig_info;
      cms->set_diag_proc_info (dpi);
    }
}

RCS_EXPORT NML_SUPER_SERVER *NML_Default_Super_Server =
  (NML_SUPER_SERVER *) NULL;

NML_SUPER_SERVER::NML_SUPER_SERVER ()
{
  servers = (RCS_LINKED_LIST *) NULL;
  unspawned_servers = 0;

  servers = new RCS_LINKED_LIST;
}

NML_SUPER_SERVER::~NML_SUPER_SERVER ()
{
  kill_all_servers ();
  delete_all_servers ();
  if (NULL != servers)
    {
      delete servers;
      servers = (RCS_LINKED_LIST *) NULL;
    }
}

void
NML_SUPER_SERVER::add_to_list (NML * _nml)
{
  NML_SERVER *server = (NML_SERVER *) NULL;
  NML_SERVER_LOCAL_PORT *local_port = (NML_SERVER_LOCAL_PORT *) NULL;
  NML *new_nml = (NML *) NULL;

  if (NULL != servers)
    {
      server = (NML_SERVER *) servers->get_head ();
      while (NULL != server)
        {
          if (server->accept_local_port_cms (_nml->cms))
            {
              break;
            }
          server = (NML_SERVER *) servers->get_next ();
        }
      if (NULL == server)
        {
          server = new NML_SERVER (_nml);
          if (NULL == server)
            {
              rcs_print_error
                ("NML_SERVER: Unable to create server object.\n");
            }
        }
      else
        {
          if (_nml->cms->isserver)
            {
              new_nml = _nml;
              local_port = new NML_SERVER_LOCAL_PORT (new_nml);
              local_port->local_channel_reused = 1;
            }
          else
            {
              new_nml = new NML (_nml, 1, -1);
              local_port = new NML_SERVER_LOCAL_PORT (new_nml);
              local_port->local_channel_reused = 0;
            }
          if (NULL == local_port)
            {
              rcs_print_error ("NML_SERVER: Unable to create local port.\n");
              return;
            }
          server->add_local_port (local_port);
        }
    }
}

void
NML_SUPER_SERVER::add_to_list (NML_SERVER * _server)
{
  if ((NULL != servers) && (NULL != _server))
    {
      _server->super_server_list_id
        = servers->store_at_tail (_server, sizeof (NML_SERVER), 0);
      unspawned_servers++;
    }
}

void
NML_SUPER_SERVER::spawn_all_servers ()
{
  NML_SERVER *server;

  if (NULL != servers)
    {
      server = (NML_SERVER *) servers->get_head ();
      while (NULL != server)
        {
          if (server->spawn () > 0 && unspawned_servers > 0)
            {
              unspawned_servers--;
            }
          server = (NML_SERVER *) servers->get_next ();
        }
    }
}

void
NML_SUPER_SERVER::kill_all_servers ()
{
  NML_SERVER *server;

  if (NULL != servers)
    {
      server = (NML_SERVER *) servers->get_head ();
      while (NULL != server)
        {
#ifdef WIN32
          DWORD pid;
          pid = GetCurrentProcessId ();
          if (server->spawner_pid != pid || pid == server->server_pid)
            {
              server = (NML_SERVER *) servers->get_next ();
              continue;
            }
#endif
#ifdef VXWORKS
          int pid;
          pid = taskIdSelf ();
          if (server->spawner_pid != pid)
            {
              server = (NML_SERVER *) servers->get_next ();
              continue;
            }
#endif
          if (server->server_spawned)
            {
              server->kill_server ();
            }
          server = (NML_SERVER *) servers->get_next ();
        }
    }
}

void
NML_SUPER_SERVER::delete_all_servers ()
{
#ifdef VXWORKS
  int pid, tid;
  pid = taskIdSelf ();
  tid = 0;
#endif
#ifdef WIN32
  DWORD pid = GetCurrentProcessId ();
  DWORD tid = GetCurrentThreadId ();
#endif
  NML_SERVER *server;
  if (NULL != servers)
    {
      server = (NML_SERVER *) servers->get_head ();
      while (NULL != server)
        {
#if defined(VXWORKS) || defined(WIN32)
          if (
              (server->server_spawned
               && (server->spawner_pid != pid || server->spawner_tid != tid))
              || (!server->server_spawned
                  && (server->server_pid != pid
                      || server->server_tid != tid)))
            {
              server = (NML_SERVER *) servers->get_next ();
              continue;
            }
#endif
          if (!server->server_spawned && unspawned_servers > 0)
            {
              unspawned_servers--;
            }
          delete server;
          server = (NML_SERVER *) servers->get_next ();
        }
    }
}


int nml_control_C_caught = 0;
int nml_sigint_count = 0;
int dont_kill_servers = 0;
int dont_cleanup_servers = 0;
static int nmlsrv_last_sig = 0;

#if defined(WIN32) && !defined(gnuwin32)
static int __stdcall
catch_control_C1 (unsigned long sig)
#else
static void
catch_control_C1 (int sig)
#endif
{
  nmlsrv_last_sig = sig;
  nml_sigint_count++;
#ifndef WIN32
  signal (SIGINT, SIG_DFL);
#endif
  if (NULL != NML_Default_Super_Server)
    {
#ifdef VXWORKS
      NML_Default_Super_Server->delete_all_servers ();
      if (NULL != NML_Default_Super_Server->servers)
        {
          if (0 == NML_Default_Super_Server->servers->list_size)
            {
#endif
              delete NML_Default_Super_Server;
              NML_Default_Super_Server = (NML_SUPER_SERVER *) NULL;
#ifdef VXWORKS
            }
        }
#endif
    }
#ifdef VXWORKS
  taskLock ();
#endif
  dont_kill_servers = 1;
  dont_cleanup_servers = 1;
  nml_cleanup ();
  dont_kill_servers = 0;
  dont_cleanup_servers = 0;
#ifdef VXWORKS
  taskUnlock ();
#endif
#ifdef UNDER_CE
  ExitThread (0);
#else
  exit (0);
#endif
#ifdef WIN32
  return (0);                   // Just to please the compiler.
#endif
}                               /*  */

#if defined(WIN32) && !defined(gnuwin32)
static int __stdcall
catch_control_C2 (unsigned long sig)
#else
static void
catch_control_C2 (int sig)
#endif
{
  nmlsrv_last_sig = sig;
#ifndef WIN32
  signal (SIGINT, SIG_DFL);
#endif
  nml_control_C_caught = 1;
#if defined(WIN32) && !defined(gnuwin32)
  return (0);
#endif
}

void
run_nml_server_exit (int i)
{
#ifndef UNDER_CE
#if defined(_WINDOWS) && !defined(gnuwin32)
  MessageBox (NULL, "Exiting run_nml_servers().", "NML Message", MB_OK);
#endif
#endif
  rcs_exit (i);
}

void RCS_EXPORT
run_nml_servers ()
{
#if defined(sun4os5) || defined(__MSDOS__) && !defined(WIN32)
  atexit (nml_cleanup);
#endif
  if (NULL != NML_Default_Super_Server)
    {
      if (NML_Default_Super_Server->servers != NULL)
        {
          if (NML_Default_Super_Server->servers->list_size <
              NML_Default_Super_Server->unspawned_servers)
            {
              NML_Default_Super_Server->unspawned_servers =
                NML_Default_Super_Server->servers->list_size;
            }
          if (NML_Default_Super_Server->unspawned_servers <= 0)
            {
              rcs_print_error
                ("run_nml_servers(): No buffers without servers already spawned for them.\n");
              return;
            }
          if (NML_Default_Super_Server->unspawned_servers == 1)
            {
              NML_Default_Super_Server->unspawned_servers = 0;
              NML_SERVER *sole_server;
              sole_server =
                (NML_SERVER *) NML_Default_Super_Server->servers->get_head ();
              while (sole_server != NULL)
                {
                  if (NULL != sole_server->remote_port)
                    {
                      if (!sole_server->remote_port->running &&
                          !sole_server->server_spawned)
                        {
                          break;
                        }
                    }
                  sole_server =
                    (NML_SERVER *) NML_Default_Super_Server->servers->
                    get_next ();
                }
              if (NULL == sole_server)
                {
                  rcs_print_error
                    ("run_nml_servers() : sole_server is NULL.\n");
                  run_nml_server_exit (-1);
                }
              else
                {
#ifndef UNDER_CE
#if defined(WIN32) && !defined(gnuwin32)
                  SetConsoleCtrlHandler (catch_control_C1, TRUE);
#else
#ifdef sparcworks_sun4
                  signal (SIGINT, (void (*)(int,...)) catch_control_C1);
#else
                  signal (SIGINT, catch_control_C1);
#endif
#endif
#endif
                  sole_server->run (0);
                  run_nml_server_exit (-1);
                }
            }
          else
            {
              nml_control_C_caught = 0;
              NML_Default_Super_Server->spawn_all_servers ();
#ifndef UNDER_CE
#if defined(WIN32) && !defined(gnuwin32)
              SetConsoleCtrlHandler (catch_control_C2, TRUE);
#else
#ifdef sparcworks_sun4
              signal (SIGINT, (void (*)(int,...)) catch_control_C2);
#else
              signal (SIGINT, catch_control_C2);
#endif
#endif
#endif
              while (!nml_control_C_caught)
                esleep (2.0);
              NML_Default_Super_Server->kill_all_servers ();
              nml_cleanup ();
              run_nml_server_exit (0);
            }
        }
      else
        {
          rcs_print_error
            ("run_nml_servers(): No buffers without servers already spawned for them.\n");
        }
    }
  else
    {
      rcs_print_error
        ("run_nml_servers(): No buffers without servers already spawned for them.\n");
    }
  run_nml_server_exit (-1);
}

void RCS_EXPORT
spawn_nml_servers ()
{
  if (NULL != NML_Default_Super_Server)
    {
      NML_Default_Super_Server->spawn_all_servers ();
    }
}

void RCS_EXPORT
kill_nml_servers ()
{
  if (!dont_kill_servers)
    {
      if (NULL != NML_Default_Super_Server)
        {
          NML_Default_Super_Server->kill_all_servers ();
        }
    }
}

void RCS_EXPORT
nml_server_cleanup ()
{
  if (!dont_cleanup_servers)
    {
      if (NULL != NML_Default_Super_Server)
        {
          NML_Default_Super_Server->kill_all_servers ();
          NML_Default_Super_Server->delete_all_servers ();
#ifdef VXWORKS
          if (NULL != NML_Default_Super_Server->servers)
            {
              if (0 == NML_Default_Super_Server->servers->list_size)
                {
#endif
                  delete NML_Default_Super_Server;
                  NML_Default_Super_Server = (NML_SUPER_SERVER *) NULL;
#ifdef VXWORKS
                }
            }
#endif
        }
    }
}

---