---

udpmem.cc

Go to the documentation of this file.

// This is neccessary to avoid muliple definitions of fd_set, etc when both
// RPC via PCNFS and Windows Sockets are to be available
#ifdef USE_PCNFS
#undef USE_PCNFS
#endif


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


#ifdef EXTERN_C_STD_HEADERS
extern "C"
{
#endif

#include <stdlib.h>             /* strtod() */
#ifndef UNDER_CE
#include <errno.h>              /* errno  */
#endif
#include <string.h>             // strstr, memset,
#include <math.h>               // ceil()
#include <ctype.h>              // isdigit()

#ifdef EXTERN_C_STD_HEADERS
}
#endif

#include "udpmem.hh"

#include "udp_opts.hh"          /* SET_UDP_NODELAY */
#include "timer.hh"             /* etime() */

#include "rem_msg.hh"           /* REMOTE_CMS_READ_REQUEST_TYPE, etc. */
#include "rcs_prnt.hh"          /* rcs_print_error() */
#include "sendmsgt.h"           /* sendmsgt() */
#include "recvmsgt.h"           /* recvmsgt() */
#include "sokintrf.h"           /* load_socket_interface(), unload_socket_interface() */
                               /* dl_gethostbyname(), dl_socket(), dl_bind() */
                                // dl_WSAGetLastError()
                                // FIONREAD

#ifdef UNDER_CE
#include "rcs_ce.h"
#endif

static int number_of_udpmem_objects;

UDPMEM::UDPMEM (char *_bufline, char *_procline):
CMS (_bufline, _procline)
{
  if (load_socket_interface () < 0)
    {
      rcs_print_error ("UDPMEM: Can not load socket interface.\n");
      status = CMS_LIBRARY_UNAVAILABLE_ERROR;
      return;
    }
  init_variables ();
  send_request = 1;
  send_broadcast = 0;
  number_of_udpmem_objects++;
  polling = (((char *) NULL) != strstr (ProcessLine, "poll"));
  memset (reply_buffer, 0, 32);
  memset (request_buffer, 0, 32);
  char *broadcast_eq = 0;
  broadcast_eq = strstr (ProcessLine, "broadcast_to_svr=");
  if (NULL != broadcast_eq)
    {
      send_broadcast_addr = dl_inet_addr (broadcast_eq + 10);
      send_broadcast = 1;
    }
  else if (0 != strstr (ProcessLine, "broadcast_to_svr"))
    {
      char localhostname[80];
      if (dl_gethostname (localhostname, 80) < 0)
        {
          if (strcmp (BufferHost, "localhost") != 0)
            {
              strncpy (localhostname, BufferHost, 80);
            }
          else
            {
              strncpy (localhostname, ProcessHost, 80);
            }
        }
#ifndef VXWORKS
      dl_gethostbyname (localhostname, &broadcast_server_host_entry);
      if (NULL == broadcast_server_host_entry)
        {
          status = CMS_CONFIG_ERROR;
          rcs_print_error ("UDPMEM: Couldn't get host address for (%s).\n",
                           localhostname);
          return;
        }
#ifdef __MSDOS__
      broadcast_server_socket_address.sin_addr.s_addr =
        *((u_long *) broadcast_server_host_entry->h_addr_list[0]);
#else
      broadcast_server_socket_address.sin_addr.s_addr =
        *((int *) broadcast_server_host_entry->h_addr_list[0]);
#endif
      broadcast_server_socket_address.sin_family =
        broadcast_server_host_entry->h_addrtype;
#else
      broadcast_server_socket_address.sin_addr.s_addr =
        hostGetByName (localhostname);
      if (broadcast_server_socket_address.sin_addr.s_addr == ERROR)
        {
          rcs_print_error ("UDPMEM: Couldn't get host address for (%s).\n",
                           localhostname);
          status = CMS_CONFIG_ERROR;
        }
#endif
      broadcast_server_socket_address.sin_addr.s_addr |= dl_htonl (0xff);
      send_broadcast_addr = broadcast_server_socket_address.sin_addr.s_addr;
      send_broadcast = 1;
      rcs_print_debug (PRINT_SOCKET_CONNECT,
                       "Broadcasting to IP address %s.\n",
                       dl_inet_ntoa
                       (broadcast_server_socket_address.sin_addr));
    }

  socket_fd = 0;
#ifndef VXWORKS
  server_host_entry = NULL;
#endif
  serial_number = 0;

  /* Set up the socket address stucture. */
  memset (&server_socket_address, 0, sizeof (server_socket_address));
  server_socket_address.sin_family = AF_INET;
  server_socket_address.sin_port = dl_htons (((u_short) udp_port_number));

  /* Get the IP address of the server using it's BufferHost. */
  if (send_broadcast)
    {
      broadcast_server_socket_address.sin_family = AF_INET;
      broadcast_server_socket_address.sin_port =
        dl_htons (((u_short) udp_port_number));
      broadcast_server_socket_address.sin_addr.s_addr = send_broadcast_addr;;
    }

  int hostname_was_address = 0;
  if (BufferHost[0] >= '0' && BufferHost[0] <= '9')
    {
      server_socket_address.sin_addr.s_addr = dl_inet_addr (BufferHost);
      if (server_socket_address.sin_addr.s_addr != 0 &&
          ((long) server_socket_address.sin_addr.s_addr) != -1)
        {
          hostname_was_address = 1;
        }
    }
  if (!hostname_was_address)
    {
#ifndef VXWORKS
      dl_gethostbyname (BufferHost, &server_host_entry);
      if (NULL == server_host_entry)
        {
          status = CMS_CONFIG_ERROR;
          rcs_print_error ("UDPMEM: Couldn't get host address for (%s).\n",
                           BufferHost);
          return;
        }
#ifdef __MSDOS__
      server_socket_address.sin_addr.s_addr =
        *((u_long *) server_host_entry->h_addr_list[0]);
#else
      server_socket_address.sin_addr.s_addr =
        *((int *) server_host_entry->h_addr_list[0]);
#endif
      server_socket_address.sin_family = server_host_entry->h_addrtype;
#else
      server_socket_address.sin_addr.s_addr = hostGetByName (BufferHost);
      if (server_socket_address.sin_addr.s_addr == ERROR)
        {
          rcs_print_error ("UDPMEM: Couldn't get host address for (%s).\n",
                           BufferHost);
          status = CMS_CONFIG_ERROR;
        }
#endif
    }
  rcs_print_debug (PRINT_SOCKET_CONNECT,
                   "Using server on %s with IP address %s.\n", BufferHost,
                   dl_inet_ntoa (server_socket_address.sin_addr));
  socket_fd = dl_socket (AF_INET, SOCK_DGRAM, 0);
  if (socket_fd < 0)
    {
#ifndef UNDER_CE
      rcs_print_error ("UDPMEM: Error from socket() (errno = %d:%s)\n",
                       errno, strerror (errno));
#endif
      status = CMS_CREATE_ERROR;
      return;
    }
  if (set_udp_socket_options (socket_fd) < 0)
    {
      status = CMS_MISC_ERROR;
      return;
    }
  char *sub_info_string = NULL;
  poll_interval_millis = 30000;
  subscription_type = CMS_NO_SUBSCRIPTION;
  cli_addr.sin_port = dl_htons (0);

  if ((min_compatible_version > 3.13 || min_compatible_version < 1e-6)
      && !send_broadcast)
    {
      sub_info_string = strstr (ProcessLine, "sub=");
      if (NULL != sub_info_string)
        {
          if (!strncmp (sub_info_string + 4, "none", 4))
            {
              subscription_type = CMS_NO_SUBSCRIPTION;
            }
          else if (!strncmp (sub_info_string + 4, "var", 3))
            {
              subscription_type = CMS_VARIABLE_SUBSCRIPTION;
            }
          else
            {
#ifndef UNDER_CE
              poll_interval_millis =
                ((int) (atof (sub_info_string + 4) * 1000.0));
#else
              poll_interval_millis =
                ((int) (RCS_CE_ATOF (sub_info_string + 4) * 1000.0));
#endif
              subscription_type = CMS_POLLED_SUBSCRIPTION;
            }
        }
      if (poll_interval_millis < ceil (clk_tck () * 1000.0))
        {
          poll_interval_millis = (int) ceil (clk_tck () * 1000.0);
        }
      char *broadcast_clnt_port_eq = strstr (buflineupper, "BROADCAST_PORT=");
      if (broadcast_clnt_port_eq != NULL)
        {
          broadcast_subscriptions = 1;
#ifndef UNDER_CE
          broadcast_clnt_port = strtol (broadcast_clnt_port_eq + 15, 0, 0);
#else
          broadcast_clnt_port = atol (broadcast_clnt_port_eq + 15);
#endif
          cli_addr.sin_port = dl_htons (broadcast_clnt_port);
        }
    }

  if (send_broadcast)
    {
      if (make_udp_socket_broadcast (socket_fd) < 0)
        {
          return;
        }
    }

  cli_addr.sin_family = AF_INET;
  cli_addr.sin_addr.s_addr = dl_htonl (INADDR_ANY);
  if (dl_bind (socket_fd, (struct sockaddr *) &cli_addr, sizeof (cli_addr)) <
      0)
    {
#if defined(_Windows) && !defined(USE_PCNFS) && !defined(gnuwin32)
      rcs_print_error ("UDPMEM: bind error %d\n", dl_WSAGetLastError ());
#else
      rcs_print_error ("UDPMEM: bind error %d = %s\n", errno,
                       strerror (errno));
#endif
    }
  memset (&request_message_header, 0, sizeof (request_message_header));
  request_message_header.msg_name = (caddr_t) & server_socket_address;
  request_message_header.msg_namelen = sizeof (server_socket_address);
  request_iov2[0].iov_base = request_buffer;
  request_iov2[0].iov_len = 20;
  if (min_compatible_version > 3.13 || min_compatible_version < 1e-6)
    {
      request_iov2[0].iov_len = 24;
    }
  if (min_compatible_version > 3.43 || min_compatible_version < 1e-6)
    {
      request_iov2[0].iov_len = 28;
    }
  else
    {
      total_subdivisions = 1;
    }
  request_iov2[1].iov_base = (char *) encoded_data;
  request_iov2[1].iov_len = max_encoded_message_size;
  request_message_header.msg_iov = request_iov2;
  request_message_header.msg_iovlen = 2;
  memset (&reply_message_header, 0, sizeof (reply_message_header));
  reply_message_header.msg_name = (caddr_t) & server_socket_address;
  reply_message_header.msg_namelen = sizeof (server_socket_address);
  reply_iov2[0].iov_base = reply_buffer;
  reply_iov2[0].iov_len = 20;
  reply_iov2[1].iov_base = (char *) encoded_data;
  reply_iov2[1].iov_len = max_encoded_message_size;
  reply_message_header.msg_iov = reply_iov2;
  reply_message_header.msg_iovlen = 2;
  char *retry_string;
  if ((retry_string = strstr (ProcessLine, "retry=")) != NULL)
    {
      retry_string += strlen ("retry=");
#ifndef UNDER_CE
      retry_timeout = strtod (retry_string, (char **) NULL);
#else
      retry_timeout = RCS_CE_ATOF (retry_string);
#endif
    }
  if (retry_timeout > timeout)
    retry_timeout = timeout;
  reply_size = 0;


  if ((min_compatible_version > 3.13 || min_compatible_version < 1e-6)
      && !send_broadcast)
    {
      if (verify_bufname () < 0)
        {
          status = CMS_STATUS_NOT_SET;
        }
      if (subscription_type != CMS_NO_SUBSCRIPTION)
        {
          setup_subscription ();
        }
    }

  if (polling)
    {
      make_udp_socket_nonblocking (socket_fd);
      timeout = 0;
      retry_timeout = 0;
    }
  last_reply_timed_out = 0;

}

int
UDPMEM::init_variables ()
{
  polling = 0;
  get_reply = 1;
  last_reply_timed_out = 0;
  reply_size = 20;
  request_size = 24;
  retry_timeout = 0.01;
  serial_number = 0;
  returned_serial_number = 0;
#ifndef VXWORKS
  server_host_entry = 0;
  broadcast_server_host_entry = 0;
#endif
  memset (&server_socket_address, 0, sizeof (struct sockaddr_in));
  memset (&broadcast_server_socket_address, 0, sizeof (struct sockaddr_in));
  socket_fd = -1;
  memset (request_buffer, 0, 32);
  memset (reply_buffer, 0, 32);
  memset (&request_message_header, 0, sizeof (struct msghdr));
  memset (&reply_message_header, 0, sizeof (struct msghdr));
  send_broadcast = 0;
  send_broadcast_addr = 0xFF;
  recieve_broadcast = 0;
  recieve_broadcast_port = 0;
  subscription_type = CMS_NO_SUBSCRIPTION;
  poll_interval_millis = 30;
  subscription_id = -1;
  send_request = 1;
  memset (&cli_addr, 0, sizeof (struct sockaddr_in));
  memset (&request_iov2, 0, 2 * sizeof (struct iovec));
  memset (&reply_iov2, 0, 2 * sizeof (struct iovec));
  broadcast_subscriptions = 0;
  broadcast_clnt_port = 0;
  return 0;
}



UDPMEM::~UDPMEM ()
{
  if (socket_fd > 0)
    {
      if (subscription_type != CMS_NO_SUBSCRIPTION)
        {
          cancel_subscription ();
        }

      if (delete_totally)
        {
          *((u_long *) request_buffer) =
            dl_htonl ((u_long) REMOTE_CMS_CLEAN_REQUEST_TYPE);
          *((u_long *) request_buffer + 1) =
            dl_htonl ((u_long) buffer_number);
          request_iov2[0].iov_len = 20;
          if (
              (min_compatible_version > 3.13
               || min_compatible_version < 1e-6))
            {
              request_iov2[0].iov_len = 24;
            }
          request_message_header.msg_iovlen = 1;
          sendmsgt (socket_fd, &request_message_header, 0, timeout);
        }
      dl_closesocket (socket_fd);
      socket_fd = 0;
    }
#if defined(_Windows) && !defined(USE_PCNFS) && !defined(gnuwin32)
  if (number_of_udpmem_objects == 1)
    {
      free_sendmsg_collection_buffer ();
      free_recvmsg_collection_buffer ();
      unload_socket_interface ();
    }
#endif
  number_of_udpmem_objects--;
}

CMS_STATUS
UDPMEM::read ()
{
  long message_size, id;
  get_reply = 1;
  send_request = (subscription_type == CMS_NO_SUBSCRIPTION);
  if (socket_fd <= 0)
    {
      rcs_print_error ("UDPMEM::read: Invalid socket descriptor. (%d)\n",
                       socket_fd);
      return (status = CMS_MISC_ERROR);
    }
  if (!last_reply_timed_out)
    {
      serial_number++;
    }
  *((u_long *) request_buffer) =
    dl_htonl ((u_long) REMOTE_CMS_READ_REQUEST_TYPE);
  *((u_long *) request_buffer + 1) = dl_htonl ((u_long) buffer_number);
  *((u_long *) request_buffer + 2) = dl_htonl ((u_long) serial_number);
  *((u_long *) request_buffer + 3) = dl_htonl ((u_long) CMS_READ_ACCESS);
  *((u_long *) request_buffer + 4) = dl_htonl ((u_long) in_buffer_id);
  if (total_subdivisions > 1)
    {
      *((u_long *) request_buffer + 6) =
        dl_htonl ((u_long) current_subdivision);
    }
  request_iov2[1].iov_len = 0;
  request_iov2[1].iov_base = (caddr_t) NULL;
  request_message_header.msg_iovlen = 1;
  reply_iov2[1].iov_len = max_encoded_message_size;
  reply_iov2[1].iov_base = (caddr_t) encoded_data;
  reply_message_header.msg_iovlen = 2;
  if (call_on_server () < 0)
    {
      rcs_print_error ("UDPMEM: read failed.\n");
      reply_message_header.msg_name = (caddr_t) & cli_addr;
      return (status = CMS_MISC_ERROR);
    }
  reply_message_header.msg_name = (caddr_t) & cli_addr;
  if (!last_reply_timed_out)
    {
      status = (CMS_STATUS) dl_ntohl (*((u_long *) reply_buffer + 1));
      message_size = dl_ntohl (*((u_long *) reply_buffer + 2));
      if (message_size + 20 != reply_size && !polling)
        {
          rcs_print_error ("UDPMEM: message_size+20 = %ld\n",
                           message_size + 20);
          rcs_print_error ("UDPMEM: reply_size = %ld\n", reply_size);
        }
      id = dl_ntohl (*((u_long *) reply_buffer + 3));
      header.was_read = dl_ntohl (*((u_long *) reply_buffer + 4));
      check_id (id);
    }
  else
    {
      if (polling)
        {
          status = CMS_READ_OLD;
        }
    }
  return (status);
}

CMS_STATUS
UDPMEM::blocking_read (double _blocking_timeout)
{
  long message_size, id;
  get_reply = 1;
  blocking_timeout = _blocking_timeout;
  double orig_timeout = timeout;
  timeout = blocking_timeout;
  int orig_polling = polling;
  if (polling)
    {
      polling = 0;
      make_udp_socket_blocking (socket_fd);
    }
  int orig_subscription_type = subscription_type;
  if (CMS_NO_SUBSCRIPTION == subscription_type)
    {
      subscription_type = CMS_POLLED_SUBSCRIPTION;
      if (blocking_timeout > 0.0)
        {
          poll_interval_millis = (int) (blocking_timeout * 100);
          if (poll_interval_millis < 10)
            {
              poll_interval_millis = 10;
            }
          if (poll_interval_millis > 1000)
            {
              poll_interval_millis = 1000;
            }
        }
      else
        {
          poll_interval_millis = 30;
        }
      setup_subscription ();
    }
  double orig_retry_timeout = retry_timeout;
  retry_timeout = timeout = blocking_timeout;
  send_request = 0;
  if (socket_fd <= 0)
    {
      rcs_print_error ("UDPMEM::read: Invalid socket descriptor. (%d)\n",
                       socket_fd);
      retry_timeout = orig_retry_timeout;
      return (status = CMS_MISC_ERROR);
    }
  if (!last_reply_timed_out)
    {
      serial_number++;
    }
  *((u_long *) request_buffer) =
    dl_htonl ((u_long) REMOTE_CMS_READ_REQUEST_TYPE);
  *((u_long *) request_buffer + 1) = dl_htonl ((u_long) buffer_number);
  *((u_long *) request_buffer + 2) = dl_htonl ((u_long) serial_number);
  *((u_long *) request_buffer + 3) = dl_htonl ((u_long) CMS_READ_ACCESS);
  *((u_long *) request_buffer + 4) = dl_htonl ((u_long) in_buffer_id);
  if (total_subdivisions > 1)
    {
      *((u_long *) request_buffer + 6) =
        dl_htonl ((u_long) current_subdivision);
    }
  request_iov2[1].iov_len = 0;
  request_iov2[1].iov_base = (caddr_t) NULL;
  request_message_header.msg_iovlen = 1;
  reply_iov2[1].iov_len = max_encoded_message_size;
  reply_iov2[1].iov_base = (caddr_t) encoded_data;
  reply_message_header.msg_iovlen = 2;
  status = CMS_READ_OLD;
  double start_time = 0.0;
  if (timeout > 1e-6)
    {
      start_time = etime ();
    }
  double time_diff = 0.0;
  while (status == CMS_READ_OLD && (timeout < 0.0 || time_diff > timeout))
    {
      if (call_on_server () < 0)
        {
          rcs_print_error ("UDPMEM: read failed.\n");
          timeout = orig_timeout;
          polling = orig_polling;
          retry_timeout = orig_retry_timeout;
          reply_message_header.msg_name = (caddr_t) & cli_addr;
          if (orig_subscription_type == CMS_NO_SUBSCRIPTION)
            {
              cancel_subscription ();
            }
          subscription_type = orig_subscription_type;
          return (status = CMS_MISC_ERROR);
        }
      reply_message_header.msg_name = (caddr_t) & cli_addr;
      if (!last_reply_timed_out)
        {
          status = (CMS_STATUS) dl_ntohl (*((u_long *) reply_buffer + 1));
          message_size = dl_ntohl (*((u_long *) reply_buffer + 2));
          if (message_size + 20 != reply_size && !polling)
            {
              rcs_print_error ("UDPMEM: message_size+20 = %ld\n",
                               message_size + 20);
              rcs_print_error ("UDPMEM: reply_size = %ld\n", reply_size);
            }
          id = dl_ntohl (*((u_long *) reply_buffer + 3));
          header.was_read = dl_ntohl (*((u_long *) reply_buffer + 4));
          check_id (id);
        }
      if (timeout > 1e-6)
        {
          time_diff = etime () - start_time;
        }
    }
  retry_timeout = orig_retry_timeout;
  timeout = orig_timeout;
  polling = orig_polling;
  if (orig_subscription_type == CMS_NO_SUBSCRIPTION)
    {
      cancel_subscription ();
    }
  subscription_type = orig_subscription_type;
  return (status);
}

CMS_STATUS
UDPMEM::peek ()
{
  long message_size, id;
  get_reply = 1;
  send_request = (subscription_type == CMS_NO_SUBSCRIPTION);
  if (socket_fd <= 0)
    {
      rcs_print_error ("UDPMEM::read: Invalid socket descriptor. (%d)\n",
                       socket_fd);
      return (status = CMS_MISC_ERROR);
    }
  if (!last_reply_timed_out)
    {
      serial_number++;
    }
  *((u_long *) request_buffer) =
    dl_htonl ((u_long) REMOTE_CMS_READ_REQUEST_TYPE);
  *((u_long *) request_buffer + 1) = dl_htonl ((u_long) buffer_number);
  *((u_long *) request_buffer + 2) = dl_htonl ((u_long) serial_number);
  *((u_long *) request_buffer + 3) = dl_htonl ((u_long) CMS_PEEK_ACCESS);
  *((u_long *) request_buffer + 4) = dl_htonl ((u_long) in_buffer_id);
  if (total_subdivisions > 1)
    {
      *((u_long *) request_buffer + 6) =
        dl_htonl ((u_long) current_subdivision);
    }
  request_iov2[1].iov_len = 0;
  request_iov2[1].iov_base = (caddr_t) NULL;
  request_message_header.msg_iovlen = 1;
  reply_iov2[1].iov_len = max_encoded_message_size;
  reply_iov2[1].iov_base = (caddr_t) encoded_data;
  reply_message_header.msg_iovlen = 2;
  if (call_on_server () < 0)
    {
      rcs_print_error ("UDPMEM: peek failed.\n");
      reply_message_header.msg_name = (caddr_t) & cli_addr;
      return (status = CMS_MISC_ERROR);
    }
  reply_message_header.msg_name = (caddr_t) & cli_addr;
  if (!last_reply_timed_out)
    {
      status = (CMS_STATUS) dl_ntohl (*((u_long *) reply_buffer + 1));
      message_size = dl_ntohl (*((u_long *) reply_buffer + 2));
      if (message_size + 20 != reply_size && !polling)
        {
          rcs_print_error ("UDPMEM: message_size+20 = %ld\n",
                           message_size + 20);
          rcs_print_error ("UDPMEM: reply_size = %ld\n", reply_size);
        }
      id = dl_ntohl (*((u_long *) reply_buffer + 3));
      header.was_read = dl_ntohl (*((u_long *) reply_buffer + 4));
      check_id (id);
    }
  else
    {
      if (polling)
        {
          status = CMS_READ_OLD;
        }
    }
  return (status);
}


int
UDPMEM::setup_subscription ()
{
  get_reply = 1;
  send_request = 1;
  double orig_retry_timeout = retry_timeout;
  if (retry_timeout < 0.0 || retry_timeout > (poll_interval_millis / 1000.0))
    {
      retry_timeout = (poll_interval_millis / 1000.0);
    }
  if (retry_timeout < clk_tck () * 2.0)
    {
      retry_timeout = clk_tck () * 2.0;
    }
  if (socket_fd <= 0)
    {
      rcs_print_error
        ("UDPMEM::setup_subscription: Invalid socket descriptor. (%d)\n",
         socket_fd);
      return (-1);
    }
  last_reply_timed_out = 0;
  serial_number++;
  *((u_long *) request_buffer) =
    dl_htonl ((u_long) REMOTE_CMS_SET_SUBSCRIPTION_REQUEST_TYPE);
  *((u_long *) request_buffer + 1) = dl_htonl ((u_long) buffer_number);
  *((u_long *) request_buffer + 2) = dl_htonl ((u_long) serial_number);
  *((u_long *) request_buffer + 3) = dl_htonl ((u_long) subscription_type);
  *((u_long *) request_buffer + 4) = dl_htonl ((u_long) poll_interval_millis);
  *((u_long *) request_buffer + 5) = dl_htonl ((u_long) in_buffer_id);
  if (total_subdivisions > 1)
    {
      *((u_long *) request_buffer + 6) =
        dl_htonl ((u_long) current_subdivision);
    }
  request_iov2[1].iov_len = 0;
  request_iov2[1].iov_base = (caddr_t) NULL;
  request_message_header.msg_iovlen = 1;
  reply_iov2[1].iov_len = 0;
  reply_iov2[1].iov_base = (caddr_t) 0;
  reply_message_header.msg_iovlen = 1;
  int orig_polling = polling;
  double orig_timeout = timeout;
  if (orig_timeout <= 0.0)
    {
      timeout = 5.0;
    }
  polling = 0;
  if (call_on_server () < 0)
    {
      retry_timeout = orig_retry_timeout;
      rcs_print_error ("UDPMEM: Can't setup subscription.\n");
      polling = orig_polling;
      timeout = orig_timeout;
      last_reply_timed_out = 0;
      subscription_type = CMS_NO_SUBSCRIPTION;
      subscription_id = -1;
      serial_number++;
      return (0);
    }
  retry_timeout = orig_retry_timeout;
  last_reply_timed_out = 0;
  serial_number++;
  polling = orig_polling;
  timeout = orig_timeout;
  if (!last_reply_timed_out)
    {
      long success = dl_htonl (*((u_long *) reply_buffer + 1));
      if (success == 1)
        {
          subscription_id = dl_htonl (*((u_long *) reply_buffer + 2));
        }
      else
        {
          rcs_print_error ("UDPMEM: Can't setup subscription.\n");
          subscription_type = CMS_NO_SUBSCRIPTION;
          subscription_id = -1;
        }
    }
  return (0);
}


int
UDPMEM::cancel_subscription ()
{
  get_reply = 1;
  send_request = 1;
  double orig_retry_timeout = retry_timeout;
  if (retry_timeout < 0.0 || retry_timeout > (poll_interval_millis / 1000.0))
    {
      retry_timeout = (poll_interval_millis / 1000.0);
    }
  if (retry_timeout < clk_tck () * 2.0)
    {
      retry_timeout = clk_tck () * 2.0;
    }
  if (socket_fd <= 0)
    {
      rcs_print_error
        ("UDPMEM::cancel_subscription: Invalid socket descriptor. (%d)\n",
         socket_fd);
      retry_timeout = orig_retry_timeout;
      return (-1);
    }
  make_udp_socket_blocking (socket_fd);
  last_reply_timed_out = 0;
  serial_number += 10000 / poll_interval_millis;
  serial_number += 10;
  *((u_long *) request_buffer) =
    dl_htonl ((u_long) REMOTE_CMS_CANCEL_SUBSCRIPTION_REQUEST_TYPE);
  *((u_long *) request_buffer + 1) = dl_htonl ((u_long) buffer_number);
  *((u_long *) request_buffer + 2) = dl_htonl ((u_long) serial_number);
  *((u_long *) request_buffer + 3) = dl_htonl ((u_long) subscription_id);
  if (total_subdivisions > 1)
    {
      *((u_long *) request_buffer + 6) =
        dl_htonl ((u_long) current_subdivision);
    }
  request_iov2[1].iov_len = 0;
  request_iov2[1].iov_base = (caddr_t) NULL;
  request_message_header.msg_iovlen = 1;
  reply_iov2[1].iov_len = 0;
  reply_iov2[1].iov_base = (caddr_t) 0;
  reply_message_header.msg_iovlen = 1;
  int orig_polling = polling;
  double orig_timeout = timeout;
  if (orig_timeout <= 10.0)
    {
      timeout = 10.0;
    }
  polling = 0;
  if (call_on_server () < 0)
    {
      retry_timeout = orig_retry_timeout;
      rcs_print_error ("UDPMEM: Can't cancel subscription.\n");
      polling = orig_polling;
      timeout = orig_timeout;
      last_reply_timed_out = 0;
      subscription_type = CMS_NO_SUBSCRIPTION;
      subscription_id = -1;
      serial_number++;
      return (0);
    }
  retry_timeout = orig_retry_timeout;
  last_reply_timed_out = 0;
  serial_number++;
  polling = orig_polling;
  timeout = orig_timeout;

  if (!last_reply_timed_out)
    {
      long success = dl_htonl (*((u_long *) reply_buffer + 1));
      if (success == 1)
        {
          if (subscription_id !=
              ((long) dl_htonl (*((u_long *) reply_buffer + 2))))
            {
              rcs_print_error
                ("UDPMEM: Can't cancel subscription. (Incorrect subscription id returned.)\n");
              subscription_type = CMS_NO_SUBSCRIPTION;
              subscription_id = -1;
            }
        }
      else
        {
          rcs_print_error ("UDPMEM: Can't cancel subscription.\n");
          subscription_type = CMS_NO_SUBSCRIPTION;
          subscription_id = -1;
        }
    }
  return (0);
}


int
UDPMEM::verify_bufname ()
{
  long message_size;
  get_reply = 1;
  send_request = 1;
  char bufname_from_server[80];
  if (socket_fd <= 0)
    {
      rcs_print_error ("UDPMEM::read: Invalid socket descriptor. (%d)\n",
                       socket_fd);
      status = CMS_MISC_ERROR;
      return (-1);
    }
  last_reply_timed_out = 0;
  serial_number++;
  *((u_long *) request_buffer) =
    dl_htonl ((u_long) REMOTE_CMS_GET_BUF_NAME_REQUEST_TYPE);
  *((u_long *) request_buffer + 1) = dl_htonl ((u_long) buffer_number);
  *((u_long *) request_buffer + 2) = dl_htonl ((u_long) serial_number);
  *((u_long *) request_buffer + 3) = dl_htonl ((u_long) CMS_PEEK_ACCESS);
  *((u_long *) request_buffer + 4) = dl_htonl ((u_long) in_buffer_id);
  if (total_subdivisions > 1)
    {
      *((u_long *) request_buffer + 6) =
        dl_htonl ((u_long) current_subdivision);
    }
  request_iov2[1].iov_len = 0;
  request_iov2[1].iov_base = (caddr_t) NULL;
  request_message_header.msg_iovlen = 1;
  reply_iov2[1].iov_len = 80;
  reply_iov2[1].iov_base = (caddr_t) bufname_from_server;
  reply_message_header.msg_iovlen = 2;
  int orig_polling = polling;
  double orig_timeout = timeout;
  if (orig_timeout <= 0.0)
    {
      timeout = 5.0;
    }
  polling = 0;
  if (call_on_server () < 0)
    {
      rcs_print_error
        ("UDPMEM: Failed to verify BufferName with server (Server may not be running.)\n");
      polling = orig_polling;
      timeout = orig_timeout;
      last_reply_timed_out = 0;
      serial_number++;
      status = CMS_NO_SERVER_ERROR;
      return (-1);
    }
  last_reply_timed_out = 0;
  serial_number++;
  polling = orig_polling;
  timeout = orig_timeout;
  if (!last_reply_timed_out)
    {
      status = (CMS_STATUS) dl_ntohl (*((u_long *) reply_buffer + 1));
      message_size = dl_ntohl (*((u_long *) reply_buffer + 2));
      if (message_size > 100)
        {
          rcs_print_error
            ("UDPMEM: message_size = %ld exceeds maximum for GET_BUF_NAME reply.\n",
             message_size + 20);
          status = CMS_MISC_ERROR;
          return (-1);
        }
      if (0 != strncmp (BufferName, bufname_from_server, 80))
        {
          rcs_print_error
            ("UDPMEM: Buffer name retrieved from server %s at IP address %s for buffer number %d was %s but %s was expected.\n",
             BufferHost, dl_inet_ntoa (server_socket_address.sin_addr),
             buffer_number,
             ((0 != bufname_from_server[0]) ? bufname_from_server : "blank"),
             BufferName);
          if (0 != bufname_from_server[0] && 0 != BufferName[0])
            {
              status = CMS_RESOURCE_CONFLICT_ERROR;
              return (-1);
            }
        }
    }
  else
    {
      status = CMS_NO_SERVER_ERROR;
      return (-1);
    }
  return (0);
}

void
throwaway_extra_data_on_socket (int socket_fd)
{
  int count = 0;
  if (socket_fd > 0)
    {
      struct msghdr temp_header;
      while (recvmsgtq (socket_fd, &temp_header, 0, 0.0) > 0 && count < 10)
        count++;
    }
}

int
UDPMEM::call_on_server ()
{
  double start_time, current_time;
  start_time = etime ();
  last_reply_timed_out = 0;
  double last_time_sent = 0.0;
  double time_diff = 0.0;
  int first_time = 1;
  while (1)
    {
      if (send_broadcast)
        {
          if (get_reply)
            {
              request_message_header.msg_name =
                (caddr_t) & server_socket_address;
              request_message_header.msg_namelen =
                sizeof (server_socket_address);
            }
          else
            {
              request_message_header.msg_name =
                (caddr_t) & broadcast_server_socket_address;
              request_message_header.msg_namelen =
                sizeof (broadcast_server_socket_address);
            }
        }
      if (send_request)
        {
          time_diff = etime () - last_time_sent;
          if (first_time || time_diff >= retry_timeout)
            {
              if (
                  (request_size =
                   sendmsgt (socket_fd, &request_message_header, 0,
                             retry_timeout)) < 0)
                {
                  if (!sendmsgt_timed_out)
                    return (-1);
                }
              last_time_sent = etime ();
              first_time = 0;
            }
        }
      if (!get_reply)
        {
          last_reply_timed_out = 0;
          return 0;
        }
      else
        {
          if (
              (reply_size =
               recvmsgtq (socket_fd, &reply_message_header, 0,
                          retry_timeout)) < 0)
            {
              if (!recvmsgt_timed_out)
                return (-1);
            }
          returned_serial_number = dl_ntohl (*((u_long *) reply_buffer));
          if (returned_serial_number == serial_number)
            {
              last_reply_timed_out = 0;
              throwaway_extra_data_on_socket (socket_fd);
              return 0;
            }
          // If we have a subscription, force the messages to the correct order.
          if (!send_request &&
              (returned_serial_number > serial_number ||
               (serial_number > 1000 && returned_serial_number < 500)))
            {
              serial_number = returned_serial_number;
              last_reply_timed_out = 0;
              return 0;
            }
          if (!polling)
            {
              current_time = etime ();
              if (current_time - start_time > timeout && timeout >= 1e-6)
                {
                  rcs_print_error
                    ("UDPMEM: time out error after %f seconds.\n",
                     current_time - start_time);
                  last_reply_timed_out = 1;
                  return -1;
                }
            }
          else
            {
              last_reply_timed_out = 1;
              return 0;
            }
        }
    }
  throwaway_extra_data_on_socket (socket_fd);
  last_reply_timed_out = 0;
  return (0);
}


CMS_STATUS
UDPMEM::write (void *user_data)
{
  get_reply = 0;
  send_request = 1;

  if (!force_raw)
    {
      user_data = encoded_data;
    }

  if (socket_fd <= 0)
    {
      rcs_print_error ("UDPMEM::write: Invalid socket descriptor. (%d)\n",
                       socket_fd);
      return (status = CMS_MISC_ERROR);
    }
  serial_number++;
  *((u_long *) request_buffer) =
    dl_htonl ((u_long) REMOTE_CMS_WRITE_REQUEST_TYPE);
  *((u_long *) request_buffer + 1) = dl_htonl ((u_long) buffer_number);
  *((u_long *) request_buffer + 2) = dl_htonl ((u_long) serial_number);
  *((u_long *) request_buffer + 3) = dl_htonl ((u_long) CMS_WRITE_ACCESS);
  *((u_long *) request_buffer + 4) =
    dl_htonl ((u_long) header.in_buffer_size);
  if (total_subdivisions > 1)
    {
      *((u_long *) request_buffer + 6) =
        dl_htonl ((u_long) current_subdivision);
    }
  request_iov2[1].iov_len = header.in_buffer_size;
  request_iov2[1].iov_base = (caddr_t) user_data;
  request_message_header.msg_iovlen = 2;
  reply_iov2[1].iov_len = 0;
  reply_iov2[1].iov_base = (caddr_t) NULL;
  reply_message_header.msg_iovlen = 1;
  if (call_on_server () < 0)
    {
      rcs_print_error ("UDPMEM: write failed.\n");
      return (status = CMS_MISC_ERROR);
    }
  status = CMS_WRITE_OK;
  header.was_read = 0;
  return (status);
}

CMS_STATUS
UDPMEM::write_if_read (void *user_data)
{
  get_reply = 0;
  send_request = 1;
  if (!force_raw)
    {
      user_data = encoded_data;
    }

  if (socket_fd <= 0)
    {
      rcs_print_error ("UDPMEM::write: Invalid socket descriptor. (%d)\n",
                       socket_fd);
      return (status = CMS_MISC_ERROR);
    }
  serial_number++;
  *((u_long *) request_buffer) =
    dl_htonl ((u_long) REMOTE_CMS_WRITE_REQUEST_TYPE);
  *((u_long *) request_buffer + 1) = dl_htonl ((u_long) buffer_number);
  *((u_long *) request_buffer + 2) = dl_htonl ((u_long) serial_number);
  *((u_long *) request_buffer + 3) =
    dl_htonl ((u_long) CMS_WRITE_IF_READ_ACCESS);
  *((u_long *) request_buffer + 4) =
    dl_htonl ((u_long) header.in_buffer_size);
  if (total_subdivisions > 1)
    {
      *((u_long *) request_buffer + 6) =
        dl_htonl ((u_long) current_subdivision);
    }
  request_iov2[1].iov_len = header.in_buffer_size;
  request_iov2[1].iov_base = (caddr_t) user_data;
  request_message_header.msg_iovlen = 2;
  reply_iov2[1].iov_len = 0;
  reply_iov2[1].iov_base = (caddr_t) NULL;
  reply_message_header.msg_iovlen = 1;
  if (call_on_server () < 0)
    {
      rcs_print_error ("UDPMEM: write_if_read failed.\n");
      return (status = CMS_MISC_ERROR);
    }
  status = CMS_WRITE_OK;
  header.was_read = 0;
  return (status);
}

int
UDPMEM::check_if_read ()
{
  get_reply = 1;
  send_request = 1;
  if (socket_fd <= 0)
    {
      rcs_print_error
        ("UDPMEM::check_if_read: Invalid socket descriptor. (%d)\n",
         socket_fd);
      return (status = CMS_MISC_ERROR);
    }
  serial_number++;
  *((u_long *) request_buffer) =
    dl_htonl ((u_long) REMOTE_CMS_CHECK_IF_READ_REQUEST_TYPE);
  *((u_long *) request_buffer + 1) = dl_htonl ((u_long) buffer_number);
  *((u_long *) request_buffer + 2) = dl_htonl ((u_long) serial_number);
  if (total_subdivisions > 1)
    {
      *((u_long *) request_buffer + 6) =
        dl_htonl ((u_long) current_subdivision);
    }
  request_iov2[1].iov_len = 0;
  request_iov2[1].iov_base = (caddr_t) NULL;
  request_message_header.msg_iovlen = 1;
  reply_iov2[1].iov_len = 0;
  reply_iov2[1].iov_base = (caddr_t) NULL;
  reply_message_header.msg_iovlen = 1;
  if (call_on_server () < 0)
    {
      rcs_print_error ("UDPMEM: check_if_read failed.\n");
      return (status = CMS_MISC_ERROR);
    }
  status = (CMS_STATUS) dl_ntohl (*((u_long *) reply_buffer + 1));
  header.was_read = dl_ntohl (*((u_long *) reply_buffer + 2));
  if (last_reply_timed_out)
    {
      header.was_read = 0;
    }
  return (header.was_read);
}


CMS_STATUS
UDPMEM::clear ()
{
  send_request = 1;
  get_reply = 0;
  if (socket_fd <= 0)
    {
      rcs_print_error
        ("UDPMEM::check_if_read: Invalid socket descriptor. (%d)\n",
         socket_fd);
      return (status = CMS_MISC_ERROR);
    }
  *((u_long *) request_buffer) =
    dl_htonl ((u_long) REMOTE_CMS_CLEAR_REQUEST_TYPE);
  *((u_long *) request_buffer + 1) = dl_htonl ((u_long) buffer_number);
  *((u_long *) request_buffer + 2) = dl_htonl ((u_long) serial_number);
  if (total_subdivisions > 1)
    {
      *((u_long *) request_buffer + 6) =
        dl_htonl ((u_long) current_subdivision);
    }
  request_iov2[1].iov_base = (caddr_t) NULL;
  request_message_header.msg_iovlen = 1;
  reply_iov2[1].iov_len = 0;
  reply_iov2[1].iov_base = (caddr_t) NULL;
  reply_message_header.msg_iovlen = 1;
  if (call_on_server () < 0)
    {
      return (status = CMS_MISC_ERROR);
    }
  status = (CMS_STATUS) dl_ntohl (*((u_long *) reply_buffer + 1));      /*  */
  return (status);
}

---