xref: /dports/math/octave-forge-parallel/parallel-4.0.1/src/p-connection.h

/*

Copyright (C) 2015-2019 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/>.

*/

#ifndef __OCT_PARALLEL_CONNECTION__

#define __OCT_PARALLEL_CONNECTION__

#include <octave/oct.h>
#include <octave/oct-refcount.h>

#include <fstream>
#include <string.h>

#include "gnulib-wrappers.h"

#include "minimal-load-save.h"

// at 'p' 37 bytes must be available; returns 0 on success and -1 on failure
int oct_parallel_store_unique_identifier (char *p);

class
octave_parallel_connection
{
public:

  octave_parallel_connection (const char *arg_peer_name, bool i_am_server,
                              const char *auuid)
    : peer_node_n (-1), cmd_stream (NULL), data_stream (NULL),
      connection_open (false), pseudo_connection (false),
      peer_name (arg_peer_name), nproc (0), nlocaljobs (0),
      server (i_am_server), uuid (auuid)
  {

  }

  octave_parallel_connection (bool i_am_server, const char *auuid)
    : peer_node_n (-1), cmd_stream (NULL), data_stream (NULL),
      connection_open (false), pseudo_connection (true),
      peer_name (""), nproc (0), nlocaljobs (0),
      server (i_am_server), uuid (auuid)


  {

  }

  octave_parallel_connection *insert_cmd_stream (octave_parallel_stream *s)
  {
    cmd_stream = s;

    connection_open = true;

    return this;
  }

  octave_parallel_connection *insert_data_stream (octave_parallel_stream *s)
  {
    data_stream = s;

    connection_open = true;

    return this;
  }

  void set_nproc (uint32_t anproc)
  {
    nproc = anproc;
  }

  void set_nlocaljobs (uint32_t anlocaljobs)
  {
    nlocaljobs = anlocaljobs;
  }

  uint32_t get_nproc (void)
  {
    return nproc;
  }

  uint32_t get_nlocaljobs (void)
  {
    return nlocaljobs;
  }

  octave_parallel_stream *get_cmd_stream (void) const
  {
    return cmd_stream;
  }

  octave_parallel_stream *get_data_stream (void) const
  {
    return data_stream;
  }

  int poll_for_errors (void)
  {
    if (! cmd_stream)
      {
        _p_error ("can't poll since no command stream present");

        return -1;
      }

    if (! cmd_stream->get_strbuf ()-> good ())
      {
        _p_error ("can't poll since command stream not good");

        return -1;
      }

    int ret = 0;

    uint32_t err;
    while (cmd_stream->get_strbuf ()->in_avail () > 0)
      {
        err = cmd_stream->network_recv_4byteint (err);

        if (! cmd_stream->good ())
          {
            _p_error ("error in reading error code");

            return -1;
          }

        _p_error ("An error with code %u occurred in server %s.\nSee %s:/tmp/octave_error-<remote-host-name>_%s.log for details.",
               err, peer_name.c_str (), peer_name.c_str (), uuid.c_str ());

        ret = 1;
      }

    return ret;
  }

  int wait_for_errors_or_data (void);

  const std::string &get_peer_name (void)
  {
    return peer_name;
  }

  ~octave_parallel_connection (void)
  {
    dlprintf ("pconnection destructor called\n");
    if (connection_open)
      close ();

    if (cmd_stream)
      {
        dlprintf ("pconnection destructor will delete cmd_stream\n");
        delete cmd_stream;
      }

    if (data_stream)
      {
        dlprintf ("pconnection destructor will delete data_stream\n");
        delete data_stream;
      }
  }

  int delete_data_stream (void)
  {
    if (data_stream)
      {
        data_stream->close ();

        if (data_stream->io_error ())
          {
            _p_error ("could not close data stream");

            return -1;
          }

        delete data_stream;

        data_stream = NULL;
      }

    return 0;
  }

  int close (void)
  {
    int ret = 0;

    if (connection_open)
      {
        if (cmd_stream)
          {
            if (! server)
              {
                poll_for_errors ();

                cmd_stream->network_send_string
                  ("sclose (sockets); __exit__;\n");
              }

            cmd_stream->close ();

            if (cmd_stream->io_error ())
              {
                _p_error ("could not close command stream");

                ret = -1;
              }
          }

        if (data_stream)
          {
            data_stream->close ();

            if (data_stream->io_error ())
              {
                _p_error ("could not close data stream");

                ret = -1;
              }
          }

        connection_open = false;

        return ret;
      }
    else if (! pseudo_connection)
      {
        _p_error ("connection is not open");

        return -1;
      }
    else
      return ret;
  }

  bool is_pseudo_connection (void)
  {
    return pseudo_connection;
  }

  bool is_server (void)
  {
    return server;
  }

  int connection_read_header (void)
  {
    return minimal_read_header (data_stream->get_istream (),
                                swap, flt_fmt);
  }

  int connection_read_data (octave_value &tc)
  {
    return minimal_read_data (data_stream->get_istream (), tc,
                              swap, flt_fmt);
  }

  bool is_open (void)
  {
    return connection_open;
  }

  const std::string &get_uuid (void)
  {
    return uuid;
  }

  int32_t peer_node_n; // -1 is unset, 0 is client

private:

  octave_parallel_stream *cmd_stream;

  octave_parallel_stream *data_stream;

  bool connection_open;

  bool pseudo_connection; // for indicating own node in network

  std::string peer_name;

  // number of usable processor cores in peer, or in own machine if
  // this is a pseudoconnection; num_processors() returns an unsigned
  // long int, but there is no specific transefer function implemented
  // here for long int, and such large numbers probably don't occur
  // for nproc; supposed to be available (i.e. non-zero) for all
  // connections in the clients network variable; not necessary in the
  // servers network variables
  uint32_t nproc;

  // configurable, this information can be used by the client to
  // decide how many local processes should be spawned at the server;
  // not necessary in the servers network variables
  uint32_t nlocaljobs;

  bool server;

  std::string uuid;

  // for Octaves save and load functions
  bool global;
  bool swap;
  OCTAVE__MACH_INFO::float_format flt_fmt;
  std::string doc;
};

class
octave_parallel_network
{
public:

  typedef Array<octave_parallel_connection *> connarray;

  octave_parallel_network (void) : refcount (0), closed (false)
  {

  }

  octave_parallel_network (int n)
    : connections (dim_vector (n, 1), NULL), refcount (0), closed (false)
  {

  }

  void resize (int n)
  {
    connections.resize1 (n, NULL);
  }

  void add_connection (octave_parallel_connection *connection)
  {
    connections.resize1 (connections.numel () + 1, connection);
  }

  void insert_connection (octave_parallel_connection *connection, int idx)
  {
    connections(idx) = connection;

    connection->peer_node_n = idx;
  }

  bool is_connection (int idx)
  {
    return (bool) connections(idx);
  }

  int close (void)
  {
    if (closed)
      {
        _p_error ("network already closed");

        return -1;
      }
    else
      {
        int ret = 0;

        for (int i = 0; i < connections.numel (); i++)
          if (connections(i) && connections(i)->close ())
            ret = -1;

        if (ret)
          _p_error ("could not close network");
        else
          closed = true;

        return ret;
      }
  }

  ~octave_parallel_network (void)
  {
    dlprintf ("pnetwork destructor called ...\n");
    for (int i = 0; i < connections.numel (); i++)
      {
        dlprintf ("... checks that connections(%i) is not NULL ...\n", i);
        if (connections(i))
          {
            dlprintf ("... and deletes it.\n");
            delete connections(i);
          }
      }
  }

  const connarray &get_ref (void)
  {
    refcount++;

    return connections;
  }

  int release_ref (void)
  {
    return --refcount;
  }


  bool is_closed (void)
  {
    return closed;
  }

private:

  connarray connections;

  OCTAVE__REFCOUNT<int> refcount;

  bool closed;
};

// This additional class is declared to circumvent const-ness of
// octave_value::get_rep() without a <const_cast>. It contains the
// members specific for the package, while the containing class
// octave_parallel_connections contains the stuff related to
// octave_base_value.
class
octave_parallel_connections_rep
{
  friend class octave_parallel_connections;

public:

  octave_parallel_connections_rep (octave_parallel_network *net,
                                   const char *auuid, bool i_am_server)
    : network (net), connections (net->get_ref ()), uuid (auuid),
      server (i_am_server), indexed (false)
  {
    subnet = connections;
  }

  // for indexing
  octave_parallel_connections_rep (octave_parallel_network *net,
                                   const char *auuid, bool i_am_server,
                                   octave_parallel_network::connarray &asubnet)
    : network (net), connections (net->get_ref ()), uuid (auuid),
      server (i_am_server), indexed (true)
  {
    subnet = asubnet;
  }

  ~octave_parallel_connections_rep (void)
  {
    dlprintf ("pconnections_rep destructor called, will request decr of network refcount and will check if it's <= 0 ...\n");
    int tpdebug;
    if ((tpdebug = network->release_ref ()) <= 0)
      {
        dlprintf ("... since it is %i, will delete network\n", tpdebug);
        delete network;
      }
  }

  int close (void)
  {
    return network->close ();
  }

  // function to return the _whole_ net (even if this is a subnet), to
  // make it possible to construct a new octave_value with this
  octave_parallel_network *get_whole_network (void)
  {
    return network;
  }

  // oct-functions work with this
  const octave_parallel_network::connarray &get_connections (void)
  {
    return subnet;
  }

  bool is_server (void)
  {
    return server;
  }

  const octave_parallel_network::connarray &get_all_connections (void)
  {
    return connections;
  }

private:

  octave_parallel_network *network;

  const octave_parallel_network::connarray &connections;

  octave_parallel_network::connarray subnet;

  std::string uuid;

  bool server;

  bool indexed;
};

class
octave_parallel_connections : public octave_base_value
{
public:

  void octave_parallel_register_type (void)
  {
    static bool type_registered = false;

    if (! type_registered)
      {
        register_type ();

        type_registered = true;
      }
  }

  octave_parallel_connections (octave_parallel_network *net,
                               const char *auuid, bool i_am_server)
    : rep (new octave_parallel_connections_rep (net, auuid, i_am_server))
  {
    octave_parallel_register_type ();
  }

  // for indexing
  octave_parallel_connections (octave_parallel_network *net,
                               const char *auuid, bool i_am_server,
                               octave_parallel_network::connarray &asubnet)
    : rep (new octave_parallel_connections_rep (net, auuid, i_am_server,
                                                asubnet))
  {
    octave_parallel_register_type ();
  }

  ~octave_parallel_connections (void)
  {
    dlprintf ("pconnections destructor called, deletes pconnections_rep\n");
    delete rep;
  }

  // Octave internal stuff

  octave_value do_index_op (const octave_value_list& idx,
                            bool resize_ok = false);

  octave_value_list do_multi_index_op (int, const octave_value_list& idx)
  {
    return do_index_op (idx);
  }

  // copied from ov-base-mat
  octave_value subsref (const std::string& type,
                        const std::list<octave_value_list>& idx)
  {
    octave_value retval;

    switch (type[0])
      {
      case '(':
        retval = do_index_op (idx.front ());
        break;

      case '{':
      case '.':
        {
          std::string nm = type_name ();
          error ("%s cannot be indexed with %c", nm.c_str (), type[0]);
        }
        break;

      default:
        panic_impossible ();
      }

    return retval.next_subsref (type, idx);
  }

  octave_value_list subsref (const std::string& type,
                             const std::list<octave_value_list>& idx, int)
  {
    return subsref (type, idx);
  }

  bool is_constant (void) const
  {
    return true;
  }

  bool is_defined (void) const
  {
    return true;
  }

  bool is_true (void) const
  {
    return ! rep->network->is_closed ();
  }

  dim_vector dims (void) const
  {
    return rep->subnet.dims ();
  }

  void print_raw (std::ostream& os, bool pr_as_read_syntax = false) const
  {
    octave_idx_type n = rep->connections.numel ();
    octave_idx_type sn = rep->subnet.numel ();

    indent (os);

    os << "<parallel cluster connections object> ";
    if (sn < n)
      os << "subnet with " << sn << " of ";
    os << n << " nodes";
    newline (os);

    os << "network id: " << rep->uuid.c_str ();
    newline (os);

    if (rep->network->is_closed ())
      os << "----- closed -----";
    else
      os << "----- open -----";
    newline (os);

    for (octave_idx_type i = 0; i < sn; i++)
      {
        os << rep->subnet(i)->peer_node_n << ") ";

        // if (rep->subnet(i)->is_server ())
        //   os << "[server] ";
        // else
        //   os << "[client] ";

        if (rep->subnet(i)->is_pseudo_connection ())
          os << "<local machine>";
        else
          os << rep->subnet(i)->get_peer_name ().c_str ();

        newline (os);
      }
  }

  void print (std::ostream& os, bool pr_as_read_syntax = false) const
  {
    print_raw (os);
  }
  // Octave changeset bcd71a2531d3 (Jan 31st 2014) made
  // octave_base_value::print() non-const, after that this virtual
  // function is not re-defined by the above print() function.  Having
  // both const and non-const print() here seems to work both with
  // Octave < and >= bcd71a2531d3 (print() is only called over the
  // parent class virtual function).
  void print (std::ostream& os, bool pr_as_read_syntax = false)
  {
    print_raw (os);
  }

  bool print_as_scalar (void) const { return true; }

  octave_parallel_connections_rep *get_rep (void) const
  {
    return rep;
  }

private:

  // needed by Octave for register_type(), also used here in indexing
  octave_parallel_connections (void)
  : rep (new octave_parallel_connections_rep
         (new octave_parallel_network (), "", false))
    {
      dlprintf ("pconnections default ctor called\n");
    }

  octave_parallel_connections_rep *rep;

  DECLARE_OV_TYPEID_FUNCTIONS_AND_DATA
};

#endif // __OCT_PARALLEL_CONNECTION__


/*
;;; Local Variables: ***
;;; mode: C++ ***
;;; End: ***
*/