VW::SpanningTree Class Reference

#include <spanning_tree.h>

Public Member Functions
	SpanningTree (short unsigned int port=26543, bool quiet=false)
	~SpanningTree ()
short unsigned int	BoundPort ()
void	Start ()
void	Run ()
void	Stop ()

Private Attributes
bool	m_stop
socket_t	sock
uint16_t	m_port
std::future< void > *	m_future
bool	m_quiet

Detailed Description

Definition at line 42 of file spanning_tree.h.

Constructor & Destructor Documentation

SpanningTree()

VW::SpanningTree::SpanningTree	(	short unsigned int	port = 26543,
		bool	quiet = false
	)

Definition at line 82 of file spanning_tree.cc.

References m_port, sock, and THROWERRNO.

                                                    : m_stop(false), m_port(port), m_future(nullptr), m_quiet(quiet)
{
#ifdef _WIN32
  WSAData wsaData;
  int lastError = WSAStartup(MAKEWORD(2, 2), &wsaData);
  if (lastError != 0)
    THROWERRNO("WSAStartup() returned error:" << lastError);
#endif

  sock = socket(PF_INET, SOCK_STREAM, 0);
  if (sock < 0)
    THROWERRNO("socket: ");

  int on = 1;
  if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char*)&on, sizeof(on)) < 0)
    THROWERRNO("setsockopt SO_REUSEADDR: ");

  sockaddr_in address;
  address.sin_family = AF_INET;
  address.sin_addr.s_addr = htonl(INADDR_ANY);

  address.sin_port = htons(port);
  if (::bind(sock, (sockaddr*)&address, sizeof(address)) < 0)
    THROWERRNO("bind failed for " << inet_ntoa(address.sin_addr));

  sockaddr_in bound_addr;
  memset(&bound_addr, 0, sizeof(bound_addr));
  socklen_t len = sizeof(bound_addr);
  if (::getsockname(sock, (sockaddr*)&bound_addr, &len) < 0)
    THROWERRNO("getsockname: " << inet_ntoa(bound_addr.sin_addr));

  // which port did we bind too (if m_port is 0 this will give us the actual port)
  m_port = ntohs(bound_addr.sin_port);
}

~SpanningTree()

VW::SpanningTree::~SpanningTree

(

)

Definition at line 117 of file spanning_tree.cc.

References m_future, and Stop().

{
  Stop();
  delete m_future;
}

Member Function Documentation

BoundPort()

short unsigned int VW::SpanningTree::BoundPort

(

)

Definition at line 123 of file spanning_tree.cc.

References m_port.

{ return m_port; }

Run()

void VW::SpanningTree::Run

(

)

Definition at line 152 of file spanning_tree.cc.

References build_tree(), client::client_ip, CLOSESOCK, f, fail_send(), partial::filled, id(), m_port, m_quiet, m_stop, partial::nodes, sock, client::socket, socket_sort(), THROW, and THROWERRNO.

Referenced by Start().

{
  std::map<size_t, partial> partial_nodesets;
  while (!m_stop)
  {
    if (listen(sock, 1024) < 0)
      THROWERRNO("listen: ");

    sockaddr_in client_address;
    socklen_t size = sizeof(client_address);
    socket_t f = accept(sock, (sockaddr*)&client_address, &size);
#ifdef _WIN32
    if (f == INVALID_SOCKET)
    {
#else
    if (f < 0)
    {
#endif
      break;
    }

    char dotted_quad[INET_ADDRSTRLEN];
    if (NULL == inet_ntop(AF_INET, &(client_address.sin_addr), dotted_quad, INET_ADDRSTRLEN))
      THROWERRNO("inet_ntop: ");

    char hostname[NI_MAXHOST];
    char servInfo[NI_MAXSERV];
    if (getnameinfo((sockaddr*)&client_address, sizeof(sockaddr), hostname, NI_MAXHOST, servInfo, NI_MAXSERV, 0))
      THROWERRNO("getnameinfo: ");

    if (!m_quiet)
      std::cerr << "inbound connection from " << dotted_quad << "(" << hostname << ':' << ntohs(m_port)
                << ") serv=" << servInfo << std::endl;

    size_t nonce = 0;
    if (recv(f, (char*)&nonce, sizeof(nonce), 0) != sizeof(nonce))
    {
      THROW(dotted_quad << "(" << hostname << ':' << ntohs(m_port) << "): nonce read failed, exiting");
    }
    else
    {
      if (!m_quiet)
        std::cerr << dotted_quad << "(" << hostname << ':' << ntohs(m_port) << "): nonce=" << nonce << std::endl;
    }
    size_t total = 0;
    if (recv(f, (char*)&total, sizeof(total), 0) != sizeof(total))
    {
      THROW(dotted_quad << "(" << hostname << ':' << ntohs(m_port) << "): total node count read failed, exiting");
    }
    else
    {
      if (!m_quiet)
        std::cerr << dotted_quad << "(" << hostname << ':' << ntohs(m_port) << "): total=" << total << std::endl;
    }
    size_t id = 0;
    if (recv(f, (char*)&id, sizeof(id), 0) != sizeof(id))
    {
      THROW(dotted_quad << "(" << hostname << ':' << ntohs(m_port) << "): node id read failed, exiting");
    }
    else
    {
      if (!m_quiet)
        std::cerr << dotted_quad << "(" << hostname << ':' << ntohs(m_port) << "): node id=" << id << std::endl;
    }

    int ok = true;
    if (id >= total)
    {
      if (!m_quiet)
        std::cout << dotted_quad << "(" << hostname << ':' << ntohs(m_port) << "): invalid id=" << id
                  << " >=  " << total << " !" << std::endl;
      ok = false;
    }
    partial partial_nodeset;

    if (partial_nodesets.find(nonce) == partial_nodesets.end())
    {
      partial_nodeset.nodes = (client*)calloc(total, sizeof(client));
      for (size_t i = 0; i < total; i++) partial_nodeset.nodes[i].client_ip = (uint32_t)-1;
      partial_nodeset.filled = 0;
    }
    else
    {
      partial_nodeset = partial_nodesets[nonce];
      partial_nodesets.erase(nonce);
    }

    if (ok && partial_nodeset.nodes[id].client_ip != (uint32_t)-1)
      ok = false;
    fail_send(f, &ok, sizeof(ok));

    if (ok)
    {
      partial_nodeset.nodes[id].client_ip = client_address.sin_addr.s_addr;
      partial_nodeset.nodes[id].socket = f;
      partial_nodeset.filled++;
    }
    if (partial_nodeset.filled != total)  // Need to wait for more connections
    {
      partial_nodesets[nonce] = partial_nodeset;
      for (size_t i = 0; i < total; i++)
      {
        if (partial_nodeset.nodes[i].client_ip == (uint32_t)-1)
        {
          if (!m_quiet)
            std::cout << "nonce " << nonce << " still waiting for " << (total - partial_nodeset.filled)
                      << " nodes out of " << total << " for example node " << i << std::endl;
          break;
        }
      }
    }
    else
    {
      // Time to make the spanning tree
      qsort(partial_nodeset.nodes, total, sizeof(client), socket_sort);

      int* parent = (int*)calloc(total, sizeof(int));
      uint16_t* kid_count = (uint16_t*)calloc(total, sizeof(uint16_t));

      int root = build_tree(parent, kid_count, total, 0);
      parent[root] = -1;

      for (size_t i = 0; i < total; i++)
      {
        fail_send(partial_nodeset.nodes[i].socket, &kid_count[i], sizeof(kid_count[i]));
      }

      uint16_t* client_ports = (uint16_t*)calloc(total, sizeof(uint16_t));

      for (size_t i = 0; i < total; i++)
      {
        int done = 0;
        if (recv(partial_nodeset.nodes[i].socket, (char*)&(client_ports[i]), sizeof(client_ports[i]), 0) <
            (int)sizeof(client_ports[i]))

          if (!m_quiet)
            std::cerr << " Port read failed for node " << i << " read " << done << std::endl;
      }  // all clients have bound to their ports.

      for (size_t i = 0; i < total; i++)
      {
        if (parent[i] >= 0)
        {
          fail_send(partial_nodeset.nodes[i].socket, &partial_nodeset.nodes[parent[i]].client_ip,
              sizeof(partial_nodeset.nodes[parent[i]].client_ip));
          fail_send(partial_nodeset.nodes[i].socket, &client_ports[parent[i]], sizeof(client_ports[parent[i]]));
        }
        else
        {
          int bogus = -1;
          uint32_t bogus2 = -1;
          fail_send(partial_nodeset.nodes[i].socket, &bogus2, sizeof(bogus2));
          fail_send(partial_nodeset.nodes[i].socket, &bogus, sizeof(bogus));
        }
        CLOSESOCK(partial_nodeset.nodes[i].socket);
      }
      free(client_ports);
      free(partial_nodeset.nodes);
      free(parent);
      free(kid_count);
    }
  }

#ifdef _WIN32
  WSACleanup();
#endif
}

Start()

void VW::SpanningTree::Start

(

)

Definition at line 125 of file spanning_tree.cc.

References m_future, and Run().

{
  // launch async
  if (m_future == nullptr)
  {
    m_future = new std::future<void>;
  }

  *m_future = std::async(std::launch::async, &SpanningTree::Run, this);
}

Stop()

void VW::SpanningTree::Stop

(

)

Definition at line 136 of file spanning_tree.cc.

References CLOSESOCK, m_future, m_stop, and sock.

Referenced by ~SpanningTree().

{
  m_stop = true;
#ifndef _WIN32
  // just close won't unblock the accept
  shutdown(sock, SHUT_RD);
#endif
  CLOSESOCK(sock);

  // wait for run to stop
  if (m_future != nullptr)
  {
    m_future->get();
  }
}

Member Data Documentation

m_future

std::future<void>* VW::SpanningTree::m_future

private

Definition at line 51 of file spanning_tree.h.

Referenced by Start(), Stop(), and ~SpanningTree().

m_port

uint16_t VW::SpanningTree::m_port

private

Definition at line 47 of file spanning_tree.h.

Referenced by BoundPort(), Run(), and SpanningTree().

m_quiet

bool VW::SpanningTree::m_quiet

private

Definition at line 53 of file spanning_tree.h.

Referenced by Run().

m_stop

bool VW::SpanningTree::m_stop

private

Definition at line 45 of file spanning_tree.h.

Referenced by Run(), and Stop().

sock

socket_t VW::SpanningTree::sock

private

Definition at line 46 of file spanning_tree.h.

Referenced by Run(), SpanningTree(), and Stop().

---

The documentation for this class was generated from the following files:

• /mnt/c/w/linux/vowpal_wabbit/vowpalwabbit/spanning_tree.h
• /mnt/c/w/linux/vowpal_wabbit/vowpalwabbit/spanning_tree.cc