src/dev/net/tcp_iface.cc - public/gem5 - Git at Google

 /*
  * Copyright (c) 2015 ARM Limited
  * All rights reserved
  *
  * The license below extends only to copyright in the software and shall
  * not be construed as granting a license to any other intellectual
  * property including but not limited to intellectual property relating
  * to a hardware implementation of the functionality of the software
  * licensed hereunder.  You may use the software subject to the license
  * terms below provided that you ensure that this notice is replicated
  * unmodified and in its entirety in all distributions of the software,
  * modified or unmodified, in source code or in binary form.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions are
  * met: redistributions of source code must retain the above copyright
  * notice, this list of conditions and the following disclaimer;
  * redistributions in binary form must reproduce the above copyright
  * notice, this list of conditions and the following disclaimer in the
  * documentation and/or other materials provided with the distribution;
  * neither the name of the copyright holders nor the names of its
  * contributors may be used to endorse or promote products derived from
  * this software without specific prior written permission.
  *
  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  */

 /* @file
  * TCP stream socket based interface class implementation for dist-gem5 runs.
  */

 #include "dev/net/tcp_iface.hh"

 #include <arpa/inet.h>
 #include <netdb.h>
 #include <netinet/tcp.h>
 #include <sys/socket.h>
 #include <sys/types.h>
 #include <unistd.h>

 #include <cerrno>
 #include <cstring>
 #include <vector>

 #include "base/types.hh"
 #include "debug/DistEthernet.hh"
 #include "debug/DistEthernetCmd.hh"
 #include "sim/sim_exit.hh"

 #if defined(__FreeBSD__)
 #include <netinet/in.h>

 #endif

 // MSG_NOSIGNAL does not exists on OS X
 #if defined(__APPLE__) || defined(__MACH__)
 #ifndef MSG_NOSIGNAL
 #define MSG_NOSIGNAL SO_NOSIGPIPE
 #endif
 #endif

 using namespace std;

 std::vector<std::pair<TCPIface::NodeInfo, int> > TCPIface::nodes;
 vector<int> TCPIface::sockRegistry;
 int TCPIface::fdStatic = -1;
 bool TCPIface::anyListening = false;

 TCPIface::TCPIface(string server_name, unsigned server_port,
                    unsigned dist_rank, unsigned dist_size,
                    Tick sync_start, Tick sync_repeat,
                    EventManager *em, bool use_pseudo_op, bool is_switch,
                    int num_nodes) :
     DistIface(dist_rank, dist_size, sync_start, sync_repeat, em, use_pseudo_op,
               is_switch, num_nodes), serverName(server_name),
     serverPort(server_port), isSwitch(is_switch), listening(false)
 {
     if (is_switch && isMaster) {
         while (!listen(serverPort)) {
             DPRINTF(DistEthernet, "TCPIface(listen): Can't bind port %d\n",
                     serverPort);
             serverPort++;
         }
         inform("tcp_iface listening on port %d", serverPort);
         // Now accept the first connection requests from each compute node and
         // store the node info. The compute nodes will then wait for ack
         // messages. Ack messages will be sent by initTransport() in the
         // appropriate order to make sure that every compute node is always
         // connected to the same switch port.
         NodeInfo ni;
         for (int i = 0; i < size; i++) {
             accept();
             DPRINTF(DistEthernet, "First connection, waiting for link info\n");
             if (!recvTCP(sock, &ni, sizeof(ni)))
                 panic("Failed to receive link info");
             nodes.push_back(make_pair(ni, sock));
         }
     }
 }

 bool
 TCPIface::listen(int port)
 {
     if (listening)
         panic("Socket already listening!");

     struct sockaddr_in sockaddr;
     int ret;

     fdStatic = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
     panic_if(fdStatic < 0, "socket() failed: %s", strerror(errno));

     sockaddr.sin_family = PF_INET;
     sockaddr.sin_addr.s_addr = INADDR_ANY;
     sockaddr.sin_port = htons(port);
     // finally clear sin_zero
     memset(&sockaddr.sin_zero, 0, sizeof(sockaddr.sin_zero));
     ret = ::bind(fdStatic, (struct sockaddr *)&sockaddr, sizeof (sockaddr));

     if (ret != 0) {
         if (ret == -1 && errno != EADDRINUSE)
             panic("ListenSocket(listen): bind() failed!");
         return false;
     }

     if (::listen(fdStatic, 24) == -1) {
         if (errno != EADDRINUSE)
             panic("ListenSocket(listen): listen() failed!");

         return false;
     }

     listening = true;
     anyListening = true;
     return true;
 }

 void
 TCPIface::establishConnection()
 {
     static unsigned cur_rank = 0;
     static unsigned cur_id = 0;
     NodeInfo ni;

     if (isSwitch) {
         if (cur_id == 0) { // first connection accepted in the ctor already
             auto const &iface0 =
                 find_if(nodes.begin(), nodes.end(),
                         [](const pair<NodeInfo, int> &cn) -> bool {
                             return cn.first.rank == cur_rank;
                         });
             assert(iface0 != nodes.end());
             assert(iface0->first.distIfaceId == 0);
             sock = iface0->second;
             ni = iface0->first;
         } else { // additional connections from the same compute node
             accept();
             DPRINTF(DistEthernet, "Next connection, waiting for link info\n");
             if (!recvTCP(sock, &ni, sizeof(ni)))
                 panic("Failed to receive link info");
             assert(ni.rank == cur_rank);
             assert(ni.distIfaceId == cur_id);
         }
         inform("Link okay  (iface:%d -> (node:%d, iface:%d))",
                distIfaceId, ni.rank, ni.distIfaceId);
         if (ni.distIfaceId < ni.distIfaceNum - 1) {
             cur_id++;
         } else {
             cur_rank++;
             cur_id = 0;
         }
         // send ack
         ni.distIfaceId = distIfaceId;
         ni.distIfaceNum = distIfaceNum;
         sendTCP(sock, &ni, sizeof(ni));
     } else { // this is not a switch
         connect();
         // send link info
         ni.rank = rank;
         ni.distIfaceId = distIfaceId;
         ni.distIfaceNum = distIfaceNum;
         sendTCP(sock, &ni, sizeof(ni));
         DPRINTF(DistEthernet, "Connected, waiting for ack (distIfaceId:%d\n",
                 distIfaceId);
         if (!recvTCP(sock, &ni, sizeof(ni)))
             panic("Failed to receive ack");
         assert(ni.rank == rank);
         inform("Link okay  (iface:%d -> switch iface:%d)", distIfaceId,
                ni.distIfaceId);
     }
     sockRegistry.push_back(sock);
 }

 void
 TCPIface::accept()
 {
     struct sockaddr_in sockaddr;
     socklen_t slen = sizeof (sockaddr);
     sock = ::accept(fdStatic, (struct sockaddr *)&sockaddr, &slen);
     if (sock != -1) {
         int i = 1;
         if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (char *)&i,
                          sizeof(i)) < 0)
             warn("ListenSocket(accept): setsockopt() TCP_NODELAY failed!");
     }
 }

 void
 TCPIface::connect()
 {
     struct addrinfo addr_hint, *addr_results;
      int ret;

      string port_str = to_string(serverPort);

      sock = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
      panic_if(sock < 0, "socket() failed: %s", strerror(errno));

      int fl = 1;
      if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (char *)&fl, sizeof(fl)) < 0)
          warn("ConnectSocket(connect): setsockopt() TCP_NODELAY failed!");

      bzero(&addr_hint, sizeof(addr_hint));
      addr_hint.ai_family = AF_INET;
      addr_hint.ai_socktype = SOCK_STREAM;
      addr_hint.ai_protocol = IPPROTO_TCP;

      ret = getaddrinfo(serverName.c_str(), port_str.c_str(),
                        &addr_hint, &addr_results);
      panic_if(ret < 0, "getaddrinf() failed: %s", strerror(errno));

      DPRINTF(DistEthernet, "Connecting to %s:%s\n",
              serverName.c_str(), port_str.c_str());

      ret = ::connect(sock, (struct sockaddr *)(addr_results->ai_addr),
                      addr_results->ai_addrlen);
      panic_if(ret < 0, "connect() failed: %s", strerror(errno));

      freeaddrinfo(addr_results);
 }

 TCPIface::~TCPIface()
 {
     int M5_VAR_USED ret;

     ret = close(sock);
     assert(ret == 0);
 }

 void
 TCPIface::sendTCP(int sock, const void *buf, unsigned length)
 {
     ssize_t ret;

     ret = ::send(sock, buf, length, MSG_NOSIGNAL);
     if (ret < 0) {
         if (errno == ECONNRESET || errno == EPIPE) {
             exitSimLoop("Message server closed connection, simulation "
                         "is exiting");
         } else {
             panic("send() failed: %s", strerror(errno));
         }
     }
     panic_if(ret != length, "send() failed");
 }

 bool
 TCPIface::recvTCP(int sock, void *buf, unsigned length)
 {
     ssize_t ret;

     ret = ::recv(sock, buf, length,  MSG_WAITALL );
     if (ret < 0) {
         if (errno == ECONNRESET || errno == EPIPE)
             inform("recv(): %s", strerror(errno));
         else if (ret < 0)
             panic("recv() failed: %s", strerror(errno));
     } else if (ret == 0) {
         inform("recv(): Connection closed");
     } else if (ret != length)
         panic("recv() failed");

     return (ret == length);
 }

 void
 TCPIface::sendPacket(const Header &header, const EthPacketPtr &packet)
 {
     sendTCP(sock, &header, sizeof(header));
     sendTCP(sock, packet->data, packet->length);
 }

 void
 TCPIface::sendCmd(const Header &header)
 {
     DPRINTF(DistEthernetCmd, "TCPIface::sendCmd() type: %d\n",
             static_cast<int>(header.msgType));
     // Global commands (i.e. sync request) are always sent by the master
     // DistIface. The transfer method is simply implemented as point-to-point
     // messages for now
     for (auto s: sockRegistry)
         sendTCP(s, (void*)&header, sizeof(header));
 }

 bool
 TCPIface::recvHeader(Header &header)
 {
     bool ret = recvTCP(sock, &header, sizeof(header));
     DPRINTF(DistEthernetCmd, "TCPIface::recvHeader() type: %d ret: %d\n",
             static_cast<int>(header.msgType), ret);
     return ret;
 }

 void
 TCPIface::recvPacket(const Header &header, EthPacketPtr &packet)
 {
     packet = make_shared<EthPacketData>(header.dataPacketLength);
     bool ret = recvTCP(sock, packet->data, header.dataPacketLength);
     panic_if(!ret, "Error while reading socket");
     packet->simLength = header.simLength;
     packet->length = header.dataPacketLength;
 }

 void
 TCPIface::initTransport()
 {
     // We cannot setup the conections in the constructor because the number
     // of dist interfaces (per process) is unknown until the (simobject) init
     // phase. That information is necessary for global connection ordering.
     establishConnection();
 }
	/*
	* Copyright (c) 2015 ARM Limited
	* All rights reserved
	*
	* The license below extends only to copyright in the software and shall
	* not be construed as granting a license to any other intellectual
	* property including but not limited to intellectual property relating
	* to a hardware implementation of the functionality of the software
	* licensed hereunder. You may use the software subject to the license
	* terms below provided that you ensure that this notice is replicated
	* unmodified and in its entirety in all distributions of the software,
	* modified or unmodified, in source code or in binary form.
	*
	* Redistribution and use in source and binary forms, with or without
	* modification, are permitted provided that the following conditions are
	* met: redistributions of source code must retain the above copyright
	* notice, this list of conditions and the following disclaimer;
	* redistributions in binary form must reproduce the above copyright
	* notice, this list of conditions and the following disclaimer in the
	* documentation and/or other materials provided with the distribution;
	* neither the name of the copyright holders nor the names of its
	* contributors may be used to endorse or promote products derived from
	* this software without specific prior written permission.
	*
	* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
	* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
	* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
	* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
	* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
	* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
	* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
	* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
	* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
	* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
	* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
	*/

	/* @file
	* TCP stream socket based interface class implementation for dist-gem5 runs.
	*/

	#include "dev/net/tcp_iface.hh"

	#include <arpa/inet.h>
	#include <netdb.h>
	#include <netinet/tcp.h>
	#include <sys/socket.h>
	#include <sys/types.h>
	#include <unistd.h>

	#include <cerrno>
	#include <cstring>
	#include <vector>

	#include "base/types.hh"
	#include "debug/DistEthernet.hh"
	#include "debug/DistEthernetCmd.hh"
	#include "sim/sim_exit.hh"

	#if defined(__FreeBSD__)
	#include <netinet/in.h>

	#endif

	// MSG_NOSIGNAL does not exists on OS X
	#if defined(__APPLE__) \|\| defined(__MACH__)
	#ifndef MSG_NOSIGNAL
	#define MSG_NOSIGNAL SO_NOSIGPIPE
	#endif
	#endif

	using namespace std;

	std::vector<std::pair<TCPIface::NodeInfo, int> > TCPIface::nodes;
	vector<int> TCPIface::sockRegistry;
	int TCPIface::fdStatic = -1;
	bool TCPIface::anyListening = false;

	TCPIface::TCPIface(string server_name, unsigned server_port,
	unsigned dist_rank, unsigned dist_size,
	Tick sync_start, Tick sync_repeat,
	EventManager *em, bool use_pseudo_op, bool is_switch,
	int num_nodes) :
	DistIface(dist_rank, dist_size, sync_start, sync_repeat, em, use_pseudo_op,
	is_switch, num_nodes), serverName(server_name),
	serverPort(server_port), isSwitch(is_switch), listening(false)
	{
	if (is_switch && isMaster) {
	while (!listen(serverPort)) {
	DPRINTF(DistEthernet, "TCPIface(listen): Can't bind port %d\n",
	serverPort);
	serverPort++;
	}
	inform("tcp_iface listening on port %d", serverPort);
	// Now accept the first connection requests from each compute node and
	// store the node info. The compute nodes will then wait for ack
	// messages. Ack messages will be sent by initTransport() in the
	// appropriate order to make sure that every compute node is always
	// connected to the same switch port.
	NodeInfo ni;
	for (int i = 0; i < size; i++) {
	accept();
	DPRINTF(DistEthernet, "First connection, waiting for link info\n");
	if (!recvTCP(sock, &ni, sizeof(ni)))
	panic("Failed to receive link info");
	nodes.push_back(make_pair(ni, sock));
	}
	}
	}

	bool
	TCPIface::listen(int port)
	{
	if (listening)
	panic("Socket already listening!");

	struct sockaddr_in sockaddr;
	int ret;

	fdStatic = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
	panic_if(fdStatic < 0, "socket() failed: %s", strerror(errno));

	sockaddr.sin_family = PF_INET;
	sockaddr.sin_addr.s_addr = INADDR_ANY;
	sockaddr.sin_port = htons(port);
	// finally clear sin_zero
	memset(&sockaddr.sin_zero, 0, sizeof(sockaddr.sin_zero));
	ret = ::bind(fdStatic, (struct sockaddr *)&sockaddr, sizeof (sockaddr));

	if (ret != 0) {
	if (ret == -1 && errno != EADDRINUSE)
	panic("ListenSocket(listen): bind() failed!");
	return false;
	}

	if (::listen(fdStatic, 24) == -1) {
	if (errno != EADDRINUSE)
	panic("ListenSocket(listen): listen() failed!");

	return false;
	}

	listening = true;
	anyListening = true;
	return true;
	}

	void
	TCPIface::establishConnection()
	{
	static unsigned cur_rank = 0;
	static unsigned cur_id = 0;
	NodeInfo ni;

	if (isSwitch) {
	if (cur_id == 0) { // first connection accepted in the ctor already
	auto const &iface0 =
	find_if(nodes.begin(), nodes.end(),
	[](const pair<NodeInfo, int> &cn) -> bool {
	return cn.first.rank == cur_rank;
	});
	assert(iface0 != nodes.end());
	assert(iface0->first.distIfaceId == 0);
	sock = iface0->second;
	ni = iface0->first;
	} else { // additional connections from the same compute node
	accept();
	DPRINTF(DistEthernet, "Next connection, waiting for link info\n");
	if (!recvTCP(sock, &ni, sizeof(ni)))
	panic("Failed to receive link info");
	assert(ni.rank == cur_rank);
	assert(ni.distIfaceId == cur_id);
	}
	inform("Link okay (iface:%d -> (node:%d, iface:%d))",
	distIfaceId, ni.rank, ni.distIfaceId);
	if (ni.distIfaceId < ni.distIfaceNum - 1) {
	cur_id++;
	} else {
	cur_rank++;
	cur_id = 0;
	}
	// send ack
	ni.distIfaceId = distIfaceId;
	ni.distIfaceNum = distIfaceNum;
	sendTCP(sock, &ni, sizeof(ni));
	} else { // this is not a switch
	connect();
	// send link info
	ni.rank = rank;
	ni.distIfaceId = distIfaceId;
	ni.distIfaceNum = distIfaceNum;
	sendTCP(sock, &ni, sizeof(ni));
	DPRINTF(DistEthernet, "Connected, waiting for ack (distIfaceId:%d\n",
	distIfaceId);
	if (!recvTCP(sock, &ni, sizeof(ni)))
	panic("Failed to receive ack");
	assert(ni.rank == rank);
	inform("Link okay (iface:%d -> switch iface:%d)", distIfaceId,
	ni.distIfaceId);
	}
	sockRegistry.push_back(sock);
	}

	void
	TCPIface::accept()
	{
	struct sockaddr_in sockaddr;
	socklen_t slen = sizeof (sockaddr);
	sock = ::accept(fdStatic, (struct sockaddr *)&sockaddr, &slen);
	if (sock != -1) {
	int i = 1;
	if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (char *)&i,
	sizeof(i)) < 0)
	warn("ListenSocket(accept): setsockopt() TCP_NODELAY failed!");
	}
	}

	void
	TCPIface::connect()
	{
	struct addrinfo addr_hint, *addr_results;
	int ret;

	string port_str = to_string(serverPort);

	sock = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
	panic_if(sock < 0, "socket() failed: %s", strerror(errno));

	int fl = 1;
	if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (char *)&fl, sizeof(fl)) < 0)
	warn("ConnectSocket(connect): setsockopt() TCP_NODELAY failed!");

	bzero(&addr_hint, sizeof(addr_hint));
	addr_hint.ai_family = AF_INET;
	addr_hint.ai_socktype = SOCK_STREAM;
	addr_hint.ai_protocol = IPPROTO_TCP;

	ret = getaddrinfo(serverName.c_str(), port_str.c_str(),
	&addr_hint, &addr_results);
	panic_if(ret < 0, "getaddrinf() failed: %s", strerror(errno));

	DPRINTF(DistEthernet, "Connecting to %s:%s\n",
	serverName.c_str(), port_str.c_str());

	ret = ::connect(sock, (struct sockaddr *)(addr_results->ai_addr),
	addr_results->ai_addrlen);
	panic_if(ret < 0, "connect() failed: %s", strerror(errno));

	freeaddrinfo(addr_results);
	}

	TCPIface::~TCPIface()
	{
	int M5_VAR_USED ret;

	ret = close(sock);
	assert(ret == 0);
	}

	void
	TCPIface::sendTCP(int sock, const void *buf, unsigned length)
	{
	ssize_t ret;

	ret = ::send(sock, buf, length, MSG_NOSIGNAL);
	if (ret < 0) {
	if (errno == ECONNRESET \|\| errno == EPIPE) {
	exitSimLoop("Message server closed connection, simulation "
	"is exiting");
	} else {
	panic("send() failed: %s", strerror(errno));
	}
	}
	panic_if(ret != length, "send() failed");
	}

	bool
	TCPIface::recvTCP(int sock, void *buf, unsigned length)
	{
	ssize_t ret;

	ret = ::recv(sock, buf, length, MSG_WAITALL );
	if (ret < 0) {
	if (errno == ECONNRESET \|\| errno == EPIPE)
	inform("recv(): %s", strerror(errno));
	else if (ret < 0)
	panic("recv() failed: %s", strerror(errno));
	} else if (ret == 0) {
	inform("recv(): Connection closed");
	} else if (ret != length)
	panic("recv() failed");

	return (ret == length);
	}

	void
	TCPIface::sendPacket(const Header &header, const EthPacketPtr &packet)
	{
	sendTCP(sock, &header, sizeof(header));
	sendTCP(sock, packet->data, packet->length);
	}

	void
	TCPIface::sendCmd(const Header &header)
	{
	DPRINTF(DistEthernetCmd, "TCPIface::sendCmd() type: %d\n",
	static_cast<int>(header.msgType));
	// Global commands (i.e. sync request) are always sent by the master
	// DistIface. The transfer method is simply implemented as point-to-point
	// messages for now
	for (auto s: sockRegistry)
	sendTCP(s, (void*)&header, sizeof(header));
	}

	bool
	TCPIface::recvHeader(Header &header)
	{
	bool ret = recvTCP(sock, &header, sizeof(header));
	DPRINTF(DistEthernetCmd, "TCPIface::recvHeader() type: %d ret: %d\n",
	static_cast<int>(header.msgType), ret);
	return ret;
	}

	void
	TCPIface::recvPacket(const Header &header, EthPacketPtr &packet)
	{
	packet = make_shared<EthPacketData>(header.dataPacketLength);
	bool ret = recvTCP(sock, packet->data, header.dataPacketLength);
	panic_if(!ret, "Error while reading socket");
	packet->simLength = header.simLength;
	packet->length = header.dataPacketLength;
	}

	void
	TCPIface::initTransport()
	{
	// We cannot setup the conections in the constructor because the number
	// of dist interfaces (per process) is unknown until the (simobject) init
	// phase. That information is necessary for global connection ordering.
	establishConnection();
	}