AllReduceSockets Class Reference

#include <allreduce.h>

Inheritance diagram for AllReduceSockets:

Public Member Functions
	AllReduceSockets (std::string pspan_server, const int pport, const size_t punique_id, size_t ptotal, const size_t pnode, bool pquiet)
virtual	~AllReduceSockets ()=default
template<class T , void(*)(T &, const T &) f>
void	all_reduce (T *buffer, const size_t n)
Public Member Functions inherited from AllReduce
	AllReduce (size_t ptotal, const size_t pnode, bool pquiet=false)
virtual	~AllReduce ()=default

Private Member Functions
void	all_reduce_init ()
template<class T >
void	pass_up (char *buffer, size_t left_read_pos, size_t right_read_pos, size_t &parent_sent_pos)
template<class T , void(*)(T &, const T &) f>
void	reduce (char *buffer, const size_t n)
void	pass_down (char *buffer, const size_t parent_read_pos, size_t &children_sent_pos)
void	broadcast (char *buffer, const size_t n)
socket_t	sock_connect (const uint32_t ip, const int port)
socket_t	getsock ()

Private Attributes
node_socks	socks
std::string	span_server
int	port
size_t	unique_id

Additional Inherited Members
Public Attributes inherited from AllReduce
const size_t	total
const size_t	node
bool	quiet

Detailed Description

Definition at line 179 of file allreduce.h.

Constructor & Destructor Documentation

AllReduceSockets()

AllReduceSockets::AllReduceSockets ( std::string  pspan_server, const int  pport, const size_t  punique_id, size_t  ptotal, const size_t  pnode, bool  pquiet  )

inline

Definition at line 290 of file allreduce.h.

      : AllReduce(ptotal, pnode, pquiet), span_server(pspan_server), port(pport), unique_id(punique_id)
  {
  }

~AllReduceSockets()

virtual AllReduceSockets::~AllReduceSockets ( )

virtualdefault

Member Function Documentation

all_reduce()

template<class T , void(*)(T &, const T &) f>

void AllReduceSockets::all_reduce ( T *  buffer, const size_t  n  )

inline

Definition at line 299 of file allreduce.h.

References node_socks::current_master.

  {
    if (span_server != socks.current_master)
      all_reduce_init();
    reduce<T, f>((char*)buffer, n * sizeof(T));
    broadcast((char*)buffer, n * sizeof(T));
  }

all_reduce_init()

void AllReduceSockets::all_reduce_init ( )

private

Definition at line 113 of file allreduce_sockets.cc.

References node_socks::children, CLOSESOCK, node_socks::current_master, f, getsock(), node_socks::parent, port, AllReduce::quiet, sock_connect(), socks, span_server, THROW, THROWERRNO, AllReduce::total, and unique_id.

{
#ifdef _WIN32
  WSAData wsaData;
  int lastError = WSAStartup(MAKEWORD(2, 2), &wsaData);
  if (lastError != 0)
    THROWERRNO("WSAStartup() returned error:" << lastError);
#endif

  struct hostent* master = gethostbyname(span_server.c_str());

  if (master == nullptr)
    THROWERRNO("gethostbyname(" << span_server << ")");

  socks.current_master = span_server;

  uint32_t master_ip = *((uint32_t*)master->h_addr);

  socket_t master_sock = sock_connect(master_ip, htons(port));
  if (send(master_sock, (const char*)&unique_id, sizeof(unique_id), 0) < (int)sizeof(unique_id))
  {
    THROW("write unique_id=" << unique_id << " to span server failed");
  }
  else
  {
    if (!quiet)
      cerr << "wrote unique_id=" << unique_id << endl;
  }
  if (send(master_sock, (const char*)&total, sizeof(total), 0) < (int)sizeof(total))
  {
    THROW("write total=" << total << " to span server failed");
  }
  else
  {
    if (!quiet)
      cerr << "wrote total=" << total << endl;
  }
  if (send(master_sock, (char*)&node, sizeof(node), 0) < (int)sizeof(node))
  {
    THROW("write node=" << node << " to span server failed");
  }
  else
  {
    if (!quiet)
      cerr << "wrote node=" << node << endl;
  }
  int ok;
  if (recv(master_sock, (char*)&ok, sizeof(ok), 0) < (int)sizeof(ok))
  {
    THROW("read ok from span server failed");
  }
  else
  {
    if (!quiet)
      cerr << "read ok=" << ok << endl;
  }
  if (!ok)
    THROW("mapper already connected");

  uint16_t kid_count;
  uint16_t parent_port;
  uint32_t parent_ip;

  if (recv(master_sock, (char*)&kid_count, sizeof(kid_count), 0) < (int)sizeof(kid_count))
  {
    THROW("read kid_count from span server failed");
  }
  else
  {
    if (!quiet)
      cerr << "read kid_count=" << kid_count << endl;
  }

  socket_t sock = -1;
  short unsigned int netport = htons(26544);
  if (kid_count > 0)
  {
    sock = getsock();
    sockaddr_in address;
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = htonl(INADDR_ANY);
    address.sin_port = netport;

    bool listening = false;
    while (!listening)
    {
      if (::bind(sock, (sockaddr*)&address, sizeof(address)) < 0)
      {
#ifdef _WIN32
        if (WSAGetLastError() == WSAEADDRINUSE)
#else
        if (errno == EADDRINUSE)
#endif
        {
          netport = htons(ntohs(netport) + 1);
          address.sin_port = netport;
        }
        else
          THROWERRNO("bind");
      }
      else
      {
        if (listen(sock, kid_count) < 0)
        {
          if (!quiet)
            cerr << "listen: " << strerror(errno) << endl;
          CLOSESOCK(sock);
          sock = getsock();
        }
        else
        {
          listening = true;
        }
      }
    }
  }

  if (send(master_sock, (const char*)&netport, sizeof(netport), 0) < (int)sizeof(netport))
    THROW("write netport failed!");

  if (recv(master_sock, (char*)&parent_ip, sizeof(parent_ip), 0) < (int)sizeof(parent_ip))
  {
    THROW("read parent_ip failed!");
  }
  else
  {
    char dotted_quad[INET_ADDRSTRLEN];
    if (nullptr == inet_ntop(AF_INET, (char*)&parent_ip, dotted_quad, INET_ADDRSTRLEN))
    {
      if (!quiet)
        cerr << "read parent_ip=" << parent_ip << "(inet_ntop: " << strerror(errno) << ")" << endl;
    }
    else
    {
      if (!quiet)
        cerr << "read parent_ip=" << dotted_quad << endl;
    }
  }
  if (recv(master_sock, (char*)&parent_port, sizeof(parent_port), 0) < (int)sizeof(parent_port))
  {
    THROW("read parent_port failed!");
  }
  else
  {
    if (!quiet)
      cerr << "read parent_port=" << parent_port << endl;
  }

  CLOSESOCK(master_sock);

  if (parent_ip != (uint32_t)-1)
  {
    socks.parent = sock_connect(parent_ip, parent_port);
  }
  else
    socks.parent = -1;

  socks.children[0] = -1;
  socks.children[1] = -1;
  for (int i = 0; i < kid_count; i++)
  {
    sockaddr_in child_address;
    socklen_t size = sizeof(child_address);
    socket_t f = accept(sock, (sockaddr*)&child_address, &size);
    if (f < 0)
      THROWERRNO("accept");

    // char hostname[NI_MAXHOST];
    // char servInfo[NI_MAXSERV];
    // getnameinfo((sockaddr *) &child_address, sizeof(sockaddr), hostname, NI_MAXHOST, servInfo, NI_MAXSERV,
    // NI_NUMERICSERV); cerr << "connected to " << hostname << ':' << ntohs(port) << endl;
    socks.children[i] = f;
  }

  if (kid_count > 0)
    CLOSESOCK(sock);
}

broadcast()

void AllReduceSockets::broadcast ( char *  buffer, const size_t  n  )

private

Definition at line 313 of file allreduce_sockets.cc.

References ar_buf_size, node_socks::children, node_socks::parent, pass_down(), socks, and THROW.

{
  size_t parent_read_pos = 0;    // First unread float from parent
  size_t children_sent_pos = 0;  // First unsent float to children
  // parent_sent_pos <= left_read_pos
  // parent_sent_pos <= right_read_pos

  if (socks.parent == -1)
  {
    parent_read_pos = n;
  }
  if (socks.children[0] == -1 && socks.children[1] == -1)
    children_sent_pos = n;

  while (parent_read_pos < n || children_sent_pos < n)
  {
    pass_down(buffer, parent_read_pos, children_sent_pos);
    if (parent_read_pos >= n && children_sent_pos >= n)
      break;

    if (socks.parent != -1)
    {
      // there is data to be read from the parent
      if (parent_read_pos == n)
        THROW("I think parent has no data to send but he thinks he has");

      size_t count = std::min(ar_buf_size, n - parent_read_pos);
      int read_size = recv(socks.parent, buffer + parent_read_pos, (int)count, 0);
      if (read_size == -1)
      {
        THROW(" recv from parent: " << strerror(errno));
      }
      parent_read_pos += read_size;
    }
  }
}

getsock()

socket_t AllReduceSockets::getsock ( )

private

Definition at line 85 of file allreduce_sockets.cc.

References AllReduce::quiet, and THROWERRNO.

Referenced by all_reduce_init().

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

    // SO_REUSEADDR will allow port rebinding on Windows, causing multiple instances
    // of VW on the same machine to potentially contact the wrong tree node.
#ifndef _WIN32
  int on = 1;
  if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char*)&on, sizeof(on)) < 0)
  {
    if (!quiet)
      cerr << "setsockopt SO_REUSEADDR: " << strerror(errno) << endl;
  }
#endif

  // Enable TCP Keep Alive to prevent socket leaks
  int enableTKA = 1;
  if (setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, (char*)&enableTKA, sizeof(enableTKA)) < 0)
  {
    if (!quiet)
      cerr << "setsockopt SO_KEEPALIVE: " << strerror(errno) << endl;
  }

  return sock;
}

pass_down()

void AllReduceSockets::pass_down ( char *  buffer, const size_t  parent_read_pos, size_t &  children_sent_pos  )

private

Definition at line 291 of file allreduce_sockets.cc.

References ar_buf_size, node_socks::children, socks, and THROW.

Referenced by broadcast().

{
  size_t my_bufsize = std::min(ar_buf_size, (parent_read_pos - children_sent_pos));

  if (my_bufsize > 0)
  {
    // going to pass up this chunk of data to the children
    if (socks.children[0] != -1 &&
        send(socks.children[0], buffer + children_sent_pos, (int)my_bufsize, 0) < (int)my_bufsize)
    {
      THROW("Write to left child failed");
    }
    if (socks.children[1] != -1 &&
        send(socks.children[1], buffer + children_sent_pos, (int)my_bufsize, 0) < (int)my_bufsize)
    {
      THROW("Write to right child failed");
    }

    children_sent_pos += my_bufsize;
  }
}

pass_up()

template<class T >

void AllReduceSockets::pass_up ( char *  buffer, size_t  left_read_pos, size_t  right_read_pos, size_t &  parent_sent_pos  )

inlineprivate

Definition at line 190 of file allreduce.h.

References ar_buf_size, node_socks::parent, and THROW.

  {
    size_t my_bufsize =
        std::min(ar_buf_size, std::min(left_read_pos, right_read_pos) / sizeof(T) * sizeof(T) - parent_sent_pos);

    if (my_bufsize > 0)
    {  // going to pass up this chunk of data to the parent
      int write_size = send(socks.parent, buffer + parent_sent_pos, (int)my_bufsize, 0);
      if (write_size < 0)
        THROW("Write to parent failed " << my_bufsize << " " << write_size << " " << parent_sent_pos << " "
                                        << left_read_pos << " " << right_read_pos);

      parent_sent_pos += write_size;
    }
  }

reduce()

template<class T , void(*)(T &, const T &) f>

void AllReduceSockets::reduce ( char *  buffer, const size_t  n  )

inlineprivate

Definition at line 207 of file allreduce.h.

References ar_buf_size, node_socks::children, node_socks::parent, THROW, and THROWERRNO.

  {
    fd_set fds;
    FD_ZERO(&fds);
    if (socks.children[0] != -1)
      FD_SET(socks.children[0], &fds);
    if (socks.children[1] != -1)
      FD_SET(socks.children[1], &fds);

    socket_t max_fd = std::max(socks.children[0], socks.children[1]) + 1;
    size_t child_read_pos[2] = {0, 0};  // First unread float from left and right children
    int child_unprocessed[2] = {0, 0};  // The number of bytes sent by the child but not yet added to the buffer
    char child_read_buf[2][ar_buf_size + sizeof(T) - 1];
    size_t parent_sent_pos = 0;  // First unsent float to parent
    // parent_sent_pos <= left_read_pos
    // parent_sent_pos <= right_read_pos

    if (socks.children[0] == -1)
    {
      child_read_pos[0] = n;
    }
    if (socks.children[1] == -1)
    {
      child_read_pos[1] = n;
    }

    while (parent_sent_pos < n || child_read_pos[0] < n || child_read_pos[1] < n)
    {
      if (socks.parent != -1)
        pass_up<T>(buffer, child_read_pos[0], child_read_pos[1], parent_sent_pos);

      if (parent_sent_pos >= n && child_read_pos[0] >= n && child_read_pos[1] >= n)
        break;

      if (child_read_pos[0] < n || child_read_pos[1] < n)
      {
        if (max_fd > 0 && select((int)max_fd, &fds, nullptr, nullptr, nullptr) == -1)
          THROWERRNO("select");

        for (int i = 0; i < 2; i++)
        {
          if (socks.children[i] != -1 && FD_ISSET(socks.children[i], &fds))
          {  // there is data to be left from left child
            if (child_read_pos[i] == n)
              THROW("I think child has no data to send but he thinks he has "
                  << FD_ISSET(socks.children[0], &fds) << " " << FD_ISSET(socks.children[1], &fds));

            size_t count = std::min(ar_buf_size, n - child_read_pos[i]);
            int read_size = recv(socks.children[i], &child_read_buf[i][child_unprocessed[i]], (int)count, 0);
            if (read_size == -1)
              THROWERRNO("recv from child");

            addbufs<T, f>((T*)buffer + child_read_pos[i] / sizeof(T), (T*)child_read_buf[i],
                (child_read_pos[i] + read_size) / sizeof(T) - child_read_pos[i] / sizeof(T));

            child_read_pos[i] += read_size;
            int old_unprocessed = child_unprocessed[i];
            child_unprocessed[i] = child_read_pos[i] % (int)sizeof(T);
            for (int j = 0; j < child_unprocessed[i]; j++)
            {
              child_read_buf[i][j] =
                  child_read_buf[i][((old_unprocessed + read_size) / (int)sizeof(T)) * sizeof(T) + j];
            }

            if (child_read_pos[i] == n)  // Done reading parent
              FD_CLR(socks.children[i], &fds);
          }
          else if (socks.children[i] != -1 && child_read_pos[i] != n)
            FD_SET(socks.children[i], &fds);
        }
      }
      if (socks.parent == -1 && child_read_pos[0] == n && child_read_pos[1] == n)
        parent_sent_pos = n;
    }
  }

sock_connect()

socket_t AllReduceSockets::sock_connect ( const uint32_t  ip, const int  port  )

private

Definition at line 36 of file allreduce_sockets.cc.

References port, AllReduce::quiet, THROW, and THROWERRNO.

Referenced by all_reduce_init().

{
  socket_t sock = socket(PF_INET, SOCK_STREAM, 0);
  if (sock == -1)
    THROWERRNO("socket");

  sockaddr_in far_end;
  far_end.sin_family = AF_INET;
  far_end.sin_port = port;

  far_end.sin_addr = *(in_addr*)&ip;
  memset(&far_end.sin_zero, '\0', 8);

  {
    char dotted_quad[INET_ADDRSTRLEN];
    if (nullptr == inet_ntop(AF_INET, &(far_end.sin_addr), dotted_quad, INET_ADDRSTRLEN))
      THROWERRNO("inet_ntop");

    char hostname[NI_MAXHOST];
    char servInfo[NI_MAXSERV];
    if (getnameinfo((sockaddr*)&far_end, sizeof(sockaddr), hostname, NI_MAXHOST, servInfo, NI_MAXSERV, NI_NUMERICSERV))
      THROWERRNO("getnameinfo(" << dotted_quad << ")");

    if (!quiet)
      cerr << "connecting to " << dotted_quad << " = " << hostname << ':' << ntohs(port) << endl;
  }

  size_t count = 0;
  int ret;
  while ((ret = connect(sock, (sockaddr*)&far_end, sizeof(far_end))) == -1 && count < 100)
  {
    count++;
    std::stringstream msg;
    if (!quiet)
    {
      msg << "connect attempt " << count << " failed: " << strerror(errno);
      cerr << msg.str() << endl;
    }
#ifdef _WIN32
    Sleep(1);
#else
    sleep(1);
#endif
  }
  if (ret == -1)
    THROW("cannot connect");
  return sock;
}

Member Data Documentation

port

int AllReduceSockets::port

private

Definition at line 184 of file allreduce.h.

Referenced by all_reduce_init(), and sock_connect().

socks

node_socks AllReduceSockets::socks

private

Definition at line 182 of file allreduce.h.

Referenced by all_reduce_init(), broadcast(), and pass_down().

span_server

std::string AllReduceSockets::span_server

private

Definition at line 183 of file allreduce.h.

Referenced by all_reduce_init().

unique_id

size_t AllReduceSockets::unique_id

private

Definition at line 185 of file allreduce.h.

Referenced by all_reduce_init().

---

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

• /mnt/c/w/linux/vowpal_wabbit/vowpalwabbit/allreduce.h
• /mnt/c/w/linux/vowpal_wabbit/vowpalwabbit/allreduce_sockets.cc