| /* |
| * 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. |
| * |
| * Authors: Gabor Dozsa |
| * Mohammad Alian |
| */ |
| |
| /* @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(); |
| } |