parallel-4.0.1/src/__octave_server__.cc

/*

Copyright (C) 2002 Hayato Fujiwara <h_fujiwara@users.sourceforge.net>
Copyright (C) 2010-2020 Olaf Till <i7tiol@t-online.de>

This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program; If not, see <http://www.gnu.org/licenses/>.

*/

// PKG_ADD: autoload ("__octave_server__", "parallel_interface.oct");
// PKG_DEL: autoload ("__octave_server__", "parallel_interface.oct", "remove");

#include <octave/oct.h>

#include "config.h"

#include <octave/file-io.h>
#include <octave/sighandlers.h>
#include <octave/parse.h>

#include <sys/socket.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <iostream>
#include <sys/stat.h>
#include <sys/poll.h>
#include <errno.h>
#include <netdb.h>
#include <sys/utsname.h>
#include <netinet/in.h> // reported necessary for FreeBSD-8

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

#include "parallel-gnutls.h"

static
int assert_file (std::string &path)
{
  OCTAVE__SYS__FILE_STAT stat (path);

  if (! stat.is_reg ())
    return -1;
  else
    return 0;
}

void
reval_loop (octave_parallel_stream &cstr)
{
  dsprintf ("reval, trying to write and read dummy\n");

  // send and read the final character of the initialization protocol
  // to and from the client
  char dummy = '\n';
  cstr.get_ostream () << dummy;
  dsprintf ("reval, wrote dummy (%i)\n", dummy);
  cstr.get_istream () >> std::noskipws >> dummy;
  dsprintf ("reval, read dummy (%i)\n", dummy);

  if (! cstr.good ())
    {
      _p_error ("could not finish initialization");
      _exit (1);
    }

  bool firsttime = true;

  dsprintf ("reval, entering loop\n");
  while (true)
    {
      std::string s;

      if (firsttime)
        {
          dsprintf ("reval, setting command to 'rehash ()' at first repetition of loop\n");
          s = "rehash ()\n"; // this avoided some problems
          firsttime = false;
        }
      else
        {
          dsprintf ("reval loop, before network_recv_string\n");
          if (cstr.network_recv_string (s))
            {
              _p_error ("error reading command");
              _exit (1);
            }
          dsprintf ("reval loop, after successful network_recv_string\n");
        }

      bool err;
      int p_err;
      SET_ERR (OCTAVE__EVAL_STRING (s, false, p_err, 0), err);
      dsprintf ("reval loop, after evaluating string\n");

      uint32_t nl = 0;
      if (err)
        {
          dsprintf ("reval loop, evaluation caused error\n");
          nl = 1;
        }
      else if (p_err)
        {
          dsprintf ("reval loop, p_err was set\n");
          nl = 1;
        }
      if (nl)
        {
          dsprintf ("reval loop, before sending error indication\n");
          if (cstr.network_send_4byteint (nl, true))
            {
              _p_error ("error sending error code");
              _exit (1);
            }
          dsprintf ("reval loop, after successful sending error indication\n");
        }
      else
        {
          dsprintf ("reval loop, no error, caring for Octave command number\n");
          if (octave_completion_matches_called)
            octave_completion_matches_called = false;
          else
            OCTAVE__COMMAND_EDITOR::increment_current_command_number ();
          dsprintf ("reval loop, no error, after caring for Octave command number\n");
        }
    }
}

DEFUN_DLD (__octave_server__, args, ,
           "-*- texinfo -*-\n\
@deftypefn {Loadable Function} {} __octave_server__ ()\n\
Undocumented internal function.\n\
@end deftypefn")
{
  // Contrary to what is done in the client side function connect.cc,
  // it is not necessarily always explicitely cared here for closing
  // of sockets and freeing of allocated memory in cases of
  // error. Since this function exits or returns in case of an
  // internal error, there is sometimes no need to do this. Returning
  // (due to exceptions) should be as good as exiting, since the
  // server is started in a way that leads to exiting Octave as soon
  // as the server returns.
  //
  // Originally, it was thought that deallocation of anything which
  // uses credentials has to be carefully provided here also, since
  // the destructor of the ref-counted credentials must be called in
  // the end to zero sensitive data before deallocation. This may not
  // be true, since the OS should zero any memory page given to a new
  // process.

  octave_value_list retval;

  std::string fname ("__octave_server__");

  dsprintf ("%s: just started\n", fname.c_str ());

  // arguments: asock, dsock, use_gnutls, cafile, hostname, server_no
  if (args.length () != 6)
    {
      _p_error ("%s: six arguments required", fname.c_str ());
      _exit (1);
    }
  // file descriptor for command connection (accepted socket)
  int asock;
  CHECK_ERROR_EXIT1 (asock = args(0).int_value (),
                     "%s: could not convert argument 1 to int",
                     fname.c_str ());
  // file descriptor for data connection (socket in listening state)
  int dsock;
  CHECK_ERROR_EXIT1 (dsock = args(1).int_value (),
                     "%s: could not convert argument 2 to int",
                     fname.c_str ());
  // use TLS
  bool use_gnutls;
  CHECK_ERROR_EXIT1 (use_gnutls = args(2).bool_value (),
                     "%s: could not convert argument 3 to bool",
                     fname.c_str ());
  // custom authentication file
  std::string cafile;
  CHECK_ERROR_EXIT1 (cafile = args(3).string_value (),
                     "%s: could not convert argument 4 to string",
                     fname.c_str ());
  // hostname
  std::string hostname;
  CHECK_ERROR_EXIT1 (hostname = args(4).string_value (),
                     "%s: could not convert argument 5 to string",
                     fname.c_str ());
  // server number for debugging
  int server_no;
  CHECK_ERROR_EXIT1 (server_no = args(5).int_value (),
                     "%s: could not convert argument 6 to int",
                     fname.c_str ());

  // A negative integer might be sent as Octave data, and Octave
  // doesn't care about coding of negative integers. (I know, there
  // probably will never be a current C-compiler with something
  // different than twos complement. But the C-standard allows for
  // it.)
  if (octave_parallel_stream::signed_int_rep ())
    {
      _p_error ("This machine doesn't seem to use twos complement as negative integer representation. If you want this to be supported, please file a bug report.");

      _exit (1);
    }

  // avoid dumping octave_core if killed by a signal
  OCTAVE__FEVAL ("sigterm_dumps_octave_core", octave_value (0), 0);
  OCTAVE__FEVAL ("sighup_dumps_octave_core", octave_value (0), 0);

  dsprintf ("%i: Octave server before start of communication\n", server_no);

#ifdef HAVE_LIBGNUTLS
  struct __ccredguard
  {
    octave_parallel_gnutls_srp_client_credentials *__c;
    __ccredguard (void) : __c (NULL) { }
    ~__ccredguard (void) { if (__c && ! __c->check_ref ()) delete __c; }
    octave_parallel_gnutls_srp_client_credentials *__get (void)
    { return __c; }
    void __set (octave_parallel_gnutls_srp_client_credentials *__ic)
    { __c = __ic; }
    void __release (void) { __c = NULL; }
  }  __ccg;
  octave_parallel_gnutls_srp_client_credentials *ccred;

  struct __scredguard
  {
    octave_parallel_gnutls_srp_server_credentials *__c;
    __scredguard (void) : __c (NULL) { }
    ~__scredguard (void) { if (__c && ! __c->check_ref ()) delete __c; }
    octave_parallel_gnutls_srp_server_credentials *__get (void)
    { return __c; }
    void __set (octave_parallel_gnutls_srp_server_credentials *__ic)
    { __c = __ic; }
    void __release (void) { __c = NULL; }
  }  __scg;
  octave_parallel_gnutls_srp_server_credentials *scred;

  if (use_gnutls)
    {
      gnutls_global_init ();
#ifdef HAVE_LIBGNUTLS_EXTRA
      gnutls_global_init_extra ();  // for SRP
      parallel_gnutls_set_mem_functions ();
#endif

      dsprintf ("%i: after initializing gnutls\n", server_no);

      // generate credentials
      if (! cafile.length ())
        {
#ifdef HAVE_OCTAVE_CONFIG_FCNS
          std::string octave_home = octave::config::octave_home ();
#else
          extern std::string Voctave_home;
          std::string octave_home = Voctave_home;
#endif
          cafile = octave_home +
            OCTAVE__SYS__FILE_OPS::dir_sep_str () + "share" +
            OCTAVE__SYS__FILE_OPS::dir_sep_str () + "octave" +
            OCTAVE__SYS__FILE_OPS::dir_sep_str () +
            "parallel-srp-data" +
            OCTAVE__SYS__FILE_OPS::dir_sep_str () + "server" +
            OCTAVE__SYS__FILE_OPS::dir_sep_str () + "passwd";
          if (assert_file (cafile))
            {
              octave_value_list f_args (1);
              f_args(0) = octave_value ("prefix");
              octave_value_list f_ret;
              CHECK_ERROR_EXIT1 (f_ret = OCTAVE__FEVAL ("pkg", f_args, 1),
                                 "%s: could not get prefix from pkg",
                                 fname.c_str ());
              CHECK_ERROR_EXIT1 (cafile = f_ret(0).string_value (),
                                 "%s: could not convert output of pkg ('prefix') to string)",
                                 fname.c_str ());
              cafile = cafile + OCTAVE__SYS__FILE_OPS::dir_sep_str () +
                "parallel-srp-data" +
                OCTAVE__SYS__FILE_OPS::dir_sep_str () + "server" +
                OCTAVE__SYS__FILE_OPS::dir_sep_str () + "passwd";
              if (assert_file (cafile))
                {
                  _p_error ("%s: no regular file found at default password file paths",
                            fname.c_str ());
                  _exit (1);
                }
            }
        }
      else if (assert_file (cafile))
        {
          _p_error ("%s: no regular file found at password file path given by user",
                    fname.c_str ());
          _exit (1);
        }
      __scg.__set (scred =
                   new octave_parallel_gnutls_srp_server_credentials
                   (cafile));
      dsprintf ("%i: after generating credentials\n", server_no);
      if (! __scg.__get ()->check_cred ())
        {
          _p_error ("%s: could not create credentials",
                    fname.c_str ());
          _exit (1);
        }
    }
#endif // HAVE_LIBGNUTLS

  // determine own number of usable processor cores
  uint32_t nproc = num_processors (NPROC_CURRENT);

  // The servers command stream will not be inserted into a
  // connection object.
  octave_parallel_streambuf *cmd_strb;
#ifdef HAVE_LIBGNUTLS
  if (use_gnutls)
    {
      dsprintf ("%i: will generate gnutls streambuf\n", server_no);
      cmd_strb = new octave_parallel_gnutls_streambuf
        (asock, scred, true);
      __scg.__release ();
      dsprintf ("%i: generated gnutls streambuf\n", server_no);
    }
  else
#endif
    cmd_strb = new octave_parallel_socket_streambuf (asock, true);
  octave_parallel_stream cmd_str (cmd_strb);
  if (! cmd_str.good ())
    {
      _p_error ("could not create command stream");
      _exit (1);
    }

  uint32_t nhosts;
  cmd_str.network_recv_4byteint (nhosts);
  dsprintf ("%i: received nhosts (%i), good: %i\n", server_no, nhosts, cmd_str.good ());

  cmd_str.network_send_4byteint (nproc, true);
  dsprintf ("%i: sent nproc (%u), good: %i\n", server_no, nproc, cmd_str.good ());

  uint32_t me;
  cmd_str.network_recv_4byteint (me);
  dsprintf ("%i: received me (%i), good: %i\n", server_no, me, cmd_str.good ());

  std::string uuid;
  cmd_str.network_recv_string (uuid);
  dsprintf ("%i: received uuid (%s), good: %i\n", server_no, uuid.c_str (), cmd_str.good ());

  if (! cmd_str.good ()) // check before using the received 'nhosts'
    {
      _p_error ("communication error in initialization");
      _exit (1);
    }
  Array<std::string> hosts (dim_vector (nhosts, 1));
  for (uint32_t i = 0; i < nhosts; i++)
    {
      cmd_str.network_recv_string (hosts(i));
      dsprintf ("%i: received hostname no %i (%s)\n", server_no, i, hosts(i).c_str ());
    }

  dsprintf ("will now change name of error file\n");
  std::string errname = std::string ("/tmp/octave_error-") + hostname.c_str ()
    + "_" + uuid.c_str () + ".log";
  struct stat s_stat;
  if (stat (errname.c_str (), &s_stat) == 0)
    {
      std::string bakname ("/tmp/octave_error-");
      bakname = bakname + hostname.c_str () +
        "_" + uuid.c_str () + ".bak";
      if (rename (errname.c_str (), bakname.c_str ()))
        {
          perror ("rename");
          _exit (1);
        }
    }
  if (! freopen (errname.c_str (), "w", stderr))
    {
      perror ("freopen");
      _exit (1);
    }

  std::string directory;
  cmd_str.network_recv_string (directory);
  dsprintf ("%i: received directory (%s)\n", server_no, directory.c_str ());

#define BUFLEN 1024
  struct __pwguard
  {
    char *__pw;
    int __len;
    __pwguard (int __alen) : __pw (new char[__alen]), __len (__alen) { }
    void __free (void)
    {
      if (__pw)
        {
          memset ((void *) __pw, 0, __len);
          delete [] __pw;
          __pw = NULL;
        }
    }
    ~__pwguard (void) { __free (); }
    char *__get (void) { return __pw; }
  } __pwg (BUFLEN);
  if (use_gnutls)
    {
      cmd_str.network_recv_string (__pwg.__get (), BUFLEN);
      if (me == nhosts) // we won't need it
        __pwg.__free ();
      dsprintf ("%i: received password (%s)\n", server_no, __pwg.__get ());
    }

  if (! cmd_str.good ())
    {
      _p_error ("communication error in initialization");
      _exit (1);
    }

  // client may shut down before starting data connections (if it was
  // unable to establish all command connections)
  struct pollfd pfd[2];
  pfd[0].fd = asock;
  pfd[0].events = POLLIN | POLLHUP; // POLLHUP meaningless here?
  pfd[0].revents = 0;
  pfd[1].fd = dsock;
  pfd[1].events = POLLIN; // will be set if accept is possible
  pfd[1].revents = 0;
  if (poll (pfd, 2, -1) == -1)
    {
      _p_error ("error in poll()");
      _exit (1);
    }
  if (pfd[0].revents)
    {
      _p_error ("unexpected event at command socket");
      _exit (1);
    }

  octave_parallel_network *network;
  struct __netwguard
  {
    octave_parallel_network *__n;
    __netwguard (octave_parallel_network *__an) : __n (__an)
    { __n->get_ref (); }
    ~__netwguard (void) { if (__n->release_ref () <= 0) delete __n; }
  } __ng (network = new octave_parallel_network (nhosts + 1));

  for (uint32_t i = 0; i < me; i++)
    {
      //      recv;

      dsprintf ("%i: trying to accept data connection, %i\n", server_no, i);
      int not_connected = 1;
      for (int j = 0; j < N_CONNECT_RETRIES; j++)
        {
          struct sockaddr_in rem_addr;
          socklen_t addrlen = sizeof (rem_addr);
          int adsock = accept (dsock, (sockaddr *) &rem_addr, &addrlen);
          if (adsock == -1)
            {
              perror ("accept, data stream");
              _exit (1);
            }
          dsprintf ("server %i, host %i, retry %i, accept successful\n", server_no, i, j);
          if (addrlen > sizeof (rem_addr))
            {
              perror ("accept, data stream, address buffer to short");
              _exit (1);
            }
          struct __sockguard
          {
            __sockguard (int __isock) { __sock = __isock; }
            ~__sockguard (void) { if (__sock > -1) close (__sock); }
            void __release (void) { __sock = -1; }
            int __sock;
          } __sockg (adsock);
#define HOSTNAMEBUFLEN 257
          char peername[HOSTNAMEBUFLEN];
          if (getnameinfo ((sockaddr *) &rem_addr, addrlen,
                           (char *) peername, HOSTNAMEBUFLEN,
                           NULL, 0, 0))
            {
              _p_error ("getnameinfo returned an error");
              _exit (1);
            }

          octave_parallel_connection *conn =
            new octave_parallel_connection
            (peername, true, uuid.c_str ());

          // you don't know the position to insert the connection
          // (protecting it) yet, an exception may be thrown before
          // you know it
          struct __connguard
          {
            __connguard (octave_parallel_connection *__ipt)
            { __pt = __ipt; }
            ~__connguard (void) { if (__pt) delete __pt; }
            void __set (octave_parallel_connection *__ipt)
            { __pt = __ipt; }
            octave_parallel_connection *__pt;
            void __release (void) { __pt = NULL; }
          } __conng (conn);

#ifdef HAVE_LIBGNUTLS
          if (use_gnutls)
            {
              conn->insert_data_stream
                (new octave_parallel_stream
                 (new octave_parallel_gnutls_streambuf
                  (adsock, scred, true)));
              dsprintf ("server %i, host %i, retry %i, generated gnutls streambuf\n", server_no, i, j);
            }
          else
#endif
            conn->insert_data_stream
              (new octave_parallel_stream
               (new octave_parallel_socket_streambuf (adsock, true)));

          __sockg.__release ();

          if (! conn->get_data_stream ()->good ())
            {
              _p_error ("could not create data stream to %s", peername);
              _exit (1);
            }

          std::string duuid;
          conn->get_data_stream ()->network_recv_string (duuid);
          dsprintf ("server %i, host %i, retry %i, received uuid (%s)\n", server_no, i, j, duuid.c_str ());

          uint32_t host_n;
          conn->get_data_stream ()->network_recv_4byteint (host_n);
          dsprintf ("server %i, host %i, retry %i, received host_n (%i)\n", server_no, i, j, host_n);

          if (! conn->get_data_stream ()->good ())
            {
              _p_error ("communication error in initialization");
              _exit (1);
            }

          if (uuid.compare (duuid))
            {
              // a different call to 'connect', i.e. a different network
              conn->get_data_stream ()->network_send_4byteint (-1);
              if (conn->delete_data_stream ())
                {
                  _p_error ("could not delete data stream");
                  _exit (1);
                }
              dsprintf ("server %i, host %i, retry %i, sent result -1\n", server_no, i, j);
              sleep (1);
            }
          else if (me <= host_n || network->is_connection (host_n))
            {
              // we should never get here (since duuid == uuid)
              conn->get_data_stream ()->network_send_4byteint (-2);
              _p_error ("server %i, host %i, retry %i, internal error, unexpected host id %i, is_connection(host id): %i", server_no, i, j, host_n,
                        network->is_connection (host_n));
              _exit (1);
            }
          else
            {
              conn->get_data_stream ()->network_send_4byteint (0);
              int err = conn->connection_read_header ();
              minimal_write_header
                (conn->get_data_stream ()->get_ostream ());
              if (err || ! conn->get_data_stream ()->good ())
                {
                  _p_error ("communication error in initialization");
                  _exit (1);
                }
              network->insert_connection (conn, host_n);
              __conng.__release ();
              not_connected = 0;
              dsprintf ("server %i, host %i, retry %i, good result sent, header read, header written and data stream good, breaking\n", server_no, i, j);
              break;
            }
        }
      if (not_connected)
        {
          _p_error ("maximum number of connect retries exceeded");
          _exit (-1);
        }
    }

  close (dsock);


  // a pseudo-connection, representing the own node in the network
  octave_parallel_connection *conn =
    new octave_parallel_connection (true, uuid.c_str ());

  network->insert_connection (conn, me);
  dsprintf ("server %i inserted pseudoconnection at %i\n", server_no, me);

  // store number of available processor cores at the own machine,
  // although this is not necessary (but we have this information
  // here...)
  conn->set_nproc (nproc);

#ifdef HAVE_LIBGNUTLS
  if (use_gnutls && me < nhosts)
    {
      const char *username =
        static_cast<octave_parallel_gnutls_streambuf*>(cmd_strb)->
        server_get_username ();
      dsprintf ("server %i determined username %s from command stream, will now allocate client credentials (for data connections) with this username and password %s\n", server_no, username, __pwg.__get ());

      __ccg.__set (ccred =
                   new octave_parallel_gnutls_srp_client_credentials
                   (username, __pwg.__get ()));
    }
#endif


  for (uint32_t i = me + 1; i <= nhosts; i++)
    {
      // connect;
      dsprintf ("connect, server %i, host %i\n", server_no, i);

      struct addrinfo *ai = NULL, hints;
      memset ((void *) &hints, 0, sizeof (hints));
      hints.ai_family = AF_INET;
      hints.ai_socktype = SOCK_STREAM;
      hints.ai_protocol = 0;
      hints.ai_flags = 0;
      if (getaddrinfo (hosts(i - 1).c_str (), "12501", &hints, &ai))
        {
          _p_error ("getaddrinfo returned an error");
          _exit (1);
        }
      struct __aiguard
      {
        __aiguard (struct addrinfo *__iai) { __ai = __iai; }
        ~__aiguard (void) { if (__ai) freeaddrinfo (__ai); }
        struct addrinfo *__ai;
      } __aig (ai);

      int not_connected = 1;
      for (int j = 0; j < N_CONNECT_RETRIES; j++)
        {
          int dsock = socket (PF_INET, SOCK_STREAM, 0);
          if (dsock == -1)
            {
              perror ("socket");
              _exit (1);
            }
          struct __sockguard
          {
            __sockguard (int __isock) { __sock = __isock; }
            ~__sockguard (void) { if (__sock > -1) close (__sock); }
            void __release (void) { __sock = -1; }
            int __sock;
          } __sockg (dsock);

          if (connect (dsock, ai->ai_addr, ai->ai_addrlen) == 0)
            {
              dsprintf ("connect, server %i, host %i, retry %i, connect succesful\n", server_no, i, j);
              octave_parallel_connection *conn =
                new octave_parallel_connection
                (hosts(i - 1).c_str (), true, uuid.c_str ());
              network->insert_connection (conn, i);

#ifdef HAVE_LIBGNUTLS
              if (use_gnutls)
                {
                  conn->insert_data_stream
                    (new octave_parallel_stream
                     (new octave_parallel_gnutls_streambuf
                      (dsock, ccred, false)));
                  __ccg.__release ();
                  dsprintf ("connect, server %i, host %i, retry %i, generated gnutls streambuf\n", server_no, i, j);
                }
              else
#endif
                conn->insert_data_stream
                  (new octave_parallel_stream
                   (new octave_parallel_socket_streambuf (dsock, false)));
              __sockg.__release ();
              if (! conn->get_data_stream ()->good ())
                {
                  _p_error ("could not create data stream to %s",
                            hosts(i - 1).c_str ());
                  _exit (1);
                }

              conn->get_data_stream ()->
                network_send_string (uuid.c_str ());
              dsprintf ("connect, server %i, host %i, retry %i, uuid written (%s)\n", server_no, i, j, uuid.c_str ());

              conn->get_data_stream ()->network_send_4byteint (me,
                                                                       true);
              dsprintf ("connect, server %i, host %i, retry %i, 'me' written (%i)\n", server_no, i, j, me);

              int32_t res;
              conn->get_data_stream ()->network_recv_4byteint (res);

              if (! conn->get_data_stream ()->good ())
                {
                  _p_error ("communication error in initialization");
                  _exit (1);
                }

              if (res == -1)
                {
                  if (conn->delete_data_stream ())
                    {
                      _p_error ("could not delete data stream");
                      _exit (1);
                    }

                  dsprintf ("connect, server %i, host %i, retry %i, sleeping after bad result\n", server_no, i, j);
                  usleep (5000);
                }
              else if (res)
                {
                  _p_error ("unexpected error in remote server");
                  _exit (1);
                }
              else
                {
                  minimal_write_header
                    (conn->get_data_stream ()->get_ostream ());
                  if (conn->connection_read_header () ||
                      ! conn->get_data_stream ()->good ())
                    {
                      _p_error ("communication error in initialization");
                      _exit (1);
                    }
                  not_connected = 0;
                  dsprintf ("connect, server %i, host %i, retry %i, good result read, header written and read and datastream good, breaking\n", server_no, i, j);
                  break;
                }
            }
          else if (errno != ECONNREFUSED && errno != EINTR)
            {
              perror ("connect");
              break;
            }
          else
            usleep (5000);
        }

      if (not_connected)
        {
          _p_error ("unable to connect to %s", hosts(i - 1).c_str ());
          _exit (1);
        }
    }

  octave_parallel_connections *conns = new octave_parallel_connections
    (network, uuid.c_str (), true);
  octave_value sockets (conns);

  __pwg.__free ();

#ifdef HAVE_OCTAVE_INTERPRETER_H
  OCTAVE__INTERPRETER::the_interpreter () -> interactive (false);
#else
  interactive = false;
#endif

  // install 'sockets' as Octave variable
  OCTAVE__INTERPRETER__SYMBOL_TABLE__ASSIGN ("sockets", sockets);
  dsprintf ("'sockets' installed\n");

  int cd_ok = OCTAVE__SYS__ENV::chdir (directory.c_str ());
  if (! cd_ok)
    {
      OCTAVE__SYS__ENV::chdir ("/tmp");
      dsprintf ("performed chdir to /tmp\n");
    }
  else
    dsprintf ("performed chdir to %s\n", directory.c_str ());

  dsprintf ("calling function reval_loop\n");
  reval_loop (cmd_str); // does not return

  return retval;
}