3
\$\begingroup\$

I recently asked for some feedback on my design for a syncronous wrapper around linux sockets and poll

Based on the feedback, I created a new (hopefully more idiomatic) design.

socket.h

#pragma once
#include <cstdint>
#include <string>

class Socket {
public:
    Socket(){}
    virtual ~Socket();

    Socket(Socket &&other);
    Socket& operator=(Socket &&other);

    int getsd() const {return sd;}
    bool has_error{false};

protected:
    explicit Socket(int sd);

    constexpr static int invalid_socket = -1;
    int sd{invalid_socket};
};


class DataSocket : public Socket {

public:
    DataSocket();
    explicit DataSocket(int sd);
    std::string get_peer_name() const;

    void send(const char* data, std::size_t len);
    void recv(char* data, std::size_t len);
};


class ConnectSocket : public DataSocket {

public:
    ConnectSocket(){}
    ConnectSocket(const char* ip_addr, std::uint16_t port);

protected:
    std::uint16_t port;
    std::uint32_t addr;
};


class ListenSocket : public Socket {

public:
    ListenSocket(std::uint16_t port);

    DataSocket accept();

protected:
    static constexpr int backlog = 128;
    std::uint16_t port;
};

socket.cpp

#include <arpa/inet.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include <cstring>
#include <iostream>
#include <stdexcept>

#include "socket.h"

using std::size_t;
using std::strerror;
using std::string;
using std::runtime_error;

namespace {
    using SaddrIn = struct sockaddr_in;
    using Saddr = struct sockaddr;

    string get_name_info(const Saddr *sa) {
        char buf[128];
        if(::getnameinfo(sa, sizeof(sockaddr), buf, 128, nullptr, 0, NI_NOFQDN | NI_NAMEREQD) != 0){
            throw std::runtime_error(string("getnameinfo: ") + std::strerror(errno));
        }
        return string(buf);
    }
}

Socket::Socket(int sd) :
    sd(sd)
{
    if (sd == invalid_socket){
        throw std::logic_error("Socket created with invalid socket descriptor");
    }
}

Socket::~Socket(){
    if(sd != invalid_socket){
        ::close(sd);
    }
}


Socket::Socket(Socket &&other)
    : sd{invalid_socket}
{
    std::swap(sd, other.sd);
    std::swap(has_error, other.has_error);
}

Socket& Socket::operator=(Socket &&other){
    std::swap(sd, other.sd);
    std::swap(has_error, other.has_error);
    return *this;
}

DataSocket::DataSocket() :
    Socket(::socket(AF_INET, SOCK_STREAM, 0))
{
}

DataSocket::DataSocket(int sd) :
    Socket(sd)
{
}

std::string DataSocket::get_peer_name() const {
    Saddr saddr;
    socklen_t size = sizeof(Saddr);
    if(::getpeername(sd, &saddr, &size) == -1){
        throw std::runtime_error(string("getpeername: ") + strerror(errno));
    }
    return get_name_info(&saddr);
}

void DataSocket::send(const char* data, size_t len){
    size_t len_sent = 0;
    while(len_sent < len){
        const size_t ret = ::send(sd, data + len_sent, len - len_sent, MSG_NOSIGNAL);
        if(ret == -1u){
            switch(errno){
            case EAGAIN:
                continue;
            default:
                throw runtime_error(string("send error: ")  + strerror(errno));
            }
        }
        len_sent += ret;
    }
}


void DataSocket::recv(char *data, size_t len){
    size_t len_recv = 0;
    while(len_recv < len){
        const size_t ret = ::recv(sd, data + len_recv, len - len_recv, 0);
        if(ret == -1u){
            throw runtime_error(string("recv error: ")  + strerror(errno));
        }
        if(ret == 0){
            throw runtime_error("remote closed");
        }
        len_recv += ret;
    }
}

ConnectSocket::ConnectSocket(const char* ip_addr, std::uint16_t port) :
    DataSocket(),
    port(port)
{
    SaddrIn saddrin{};
    saddrin.sin_family = AF_INET;
    saddrin.sin_port   = ::htons(port);

    try {
        if (::inet_aton(ip_addr, &saddrin.sin_addr) == 0){
            throw runtime_error("inet_aton error on " + string(ip_addr) + ":" + std::to_string(port) + ": " + strerror(errno) );
        }

        if (::connect(sd, reinterpret_cast<Saddr*>(&saddrin), sizeof(Saddr)) != 0){
            throw runtime_error("connect error on " + string(ip_addr) + ":" + std::to_string(port) + ": " + strerror(errno));
        }
    } catch(const runtime_error &e) {
        ::close(sd);
        throw;
    }
}

ListenSocket::ListenSocket(std::uint16_t port) :
    Socket(::socket(AF_INET, SOCK_STREAM, 0))
{
    try {
        const int opt = 1;
        if(::setsockopt(sd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(int)) < 0){
            throw std::runtime_error(string("setsockopt: ") + strerror(errno));
        }

        SaddrIn saddrin{};
        saddrin.sin_family = AF_INET;
        saddrin.sin_port   = ::htons(port);
        if(::bind(sd, reinterpret_cast<Saddr*>(&saddrin), sizeof(Saddr)) != 0){
            throw std::runtime_error(string("bind: ") + strerror(errno));
        }

        if(::listen(sd, backlog) != 0){
            throw std::runtime_error(string("listen: ") + strerror(errno));
        }
    } catch (const std::exception&) {
        ::close(sd);
        throw;
    }
}

DataSocket ListenSocket::accept(){
    Saddr saddr;
    socklen_t addrlen = sizeof(Saddr);
    const int ret = ::accept(sd, &saddr, &addrlen);
    if(ret < 0){
        throw std::runtime_error(string("accept: ") + strerror(errno));
    }

    return DataSocket{ret};
}

pollservice.h

#pragma once
#include <poll.h>
#include <cstring>
#include <vector>
#include <iostream>
#include "socket.h"

//Collection of sockets, mutliplex over them with poll
class PollService {

    //Allow to use, say, either Socket* or shared_ptr<Socket>
    template<class S>
    class PollSet {
    public:
        //unlikely to have a use for a const iterator
        typedef typename std::vector<S>::iterator iterator;
        iterator begin() { return polled_sockets.begin(); }
        iterator end() { return polled_sockets.end(); }

        void add(S s) { polled_sockets.push_back(s); }
    private:
        std::vector<S> polled_sockets;
    };

public:

    PollService(int timeout_ms) : timeout_ms{timeout_ms} {}
    PollService(const PollService&) = delete;
    PollService& operator=(const PollService&) = delete;

    template<typename S, typename Iter>
    PollSet<S> poll_sockets(Iter start, Iter end);

private:
    using Pollfd = struct pollfd;

    const int timeout_ms;
    std::vector<Pollfd> pfds;
};

template<typename S, typename Iter>
PollService::PollSet<S> PollService::poll_sockets(Iter start, Iter end){
    PollSet<S> poll_set;
    pfds.clear();

    auto i = start;
    while(i != end){
        pfds.push_back({(*i)->getsd(), POLLIN, 0});
        i++;
    }

    const int ret = ::poll(&pfds[0], pfds.size(), timeout_ms);
    if(ret < 0){
        throw std::runtime_error(std::string("poll: ") + std::strerror(errno));
    }

    i = start;
    for (auto p = pfds.begin(); p != pfds.end() && i != end; ++p){
        if(p->revents & (POLLHUP | POLLERR | POLLNVAL | POLLIN)){
            for ( ; i != end && (*i)->getsd() != p->fd; ++i);
            if (i == end){
                break;
            }
            if (p->revents & (POLLHUP | POLLERR | POLLNVAL)){
                (*i)->has_error = true;
            }

            poll_set.add(&(**i));
        }
    }

    return poll_set;
}

Example usage:

main.cpp

#include "socket.h"
#include "pollservice.h"
#include <memory>
#include <iostream>
#include <algorithm>

int main(){
    std::vector<std::unique_ptr<Socket>> sockets;
    sockets.push_back(std::make_unique<ListenSocket>(33333));

    PollService poll_service{500};

    while (true){
        for (auto socket : poll_service.poll_sockets<Socket *>(sockets.begin(), sockets.end())){
            try {
                if(socket->has_error) {
                    throw std::runtime_error("bad poll");
                }

                if(ListenSocket* listen_socket = dynamic_cast<ListenSocket*>(socket)){
                    auto ds = listen_socket->accept();
                    std::cout << "accepted " << ds.get_peer_name() << std::endl;
                    sockets.push_back(std::make_unique<DataSocket>(std::move(ds)));
                } else {
                    if (auto s = dynamic_cast<DataSocket*>(socket)){
                        char buf[5];
                        s->recv(buf, 4);
                        buf[5] = '\0';
                        std::cout << buf << std::endl;
                    } else {
                        throw std::runtime_error("Couldn't cast socket to DataSocket");
                    }
                }
            } catch (const std::runtime_error &e){
                std::cerr << "caught error: " << e.what() << std::endl;
                sockets.erase(std::remove_if(sockets.begin(), sockets.end(), [socket](const std::unique_ptr<Socket> &s){
                    return socket->getsd() == s->getsd();
                }));

            }
        }
    }
}

I'm most interested in a review focusing on the pollservice. Could I improve the way I construct and return the iterator? I would also like to be able to statically constrain the type of the iterator if possible, but std::is_pointer is to constricting (doesn't allow smart pointers).

\$\endgroup\$
  • \$\begingroup\$ LGTM: Looks Good to Me. \$\endgroup\$ – Martin York Jul 18 '16 at 23:48
  • \$\begingroup\$ @LokiAstari, cheers, also thanks for your socket tutorials - which formed some of the inspiration for this code! \$\endgroup\$ – cloakedlearning Jul 19 '16 at 2:31
  • \$\begingroup\$ I noticed. :- ) \$\endgroup\$ – Martin York Jul 19 '16 at 5:19
4
\$\begingroup\$

Dynamic allocation inside the reactor loop should be avoided whenever possible (you allocate std::vector inside the PollSet). It would not only be very slow itself, but will slow the code around by increasing contention on the memory allocators and increasing heap fragmentation.

One possible option is to accept a functor instead of returning the container: void poll_sockets(auto& beg, auto& end, auto& fun) { ...; f(*itr); ... }

\$\endgroup\$
  • \$\begingroup\$ Yeah, that makes sense. Perhaps I could use a stack allocator to avoid changing my design? The vector is almost always going to contain a pointer of some variety so most likely a max of 16bytes per element. \$\endgroup\$ – cloakedlearning Jul 19 '16 at 12:03
  • \$\begingroup\$ Usually you'd want the code running in the IO loop to be as lightweight as possible as it potentially gets executed tens of millions times per second. any copying, no matter how little may slow it down noticeably. best is to write a micro benchmark on a loopback socket and see for yourself. \$\endgroup\$ – bobah Jul 19 '16 at 12:26
  • \$\begingroup\$ Yeah, I see your point. My current use case isn't for extremely high throughput sockets, so I'm thinking about making the vector thread_local for now as a compromise (to avoid most of the allocations). \$\endgroup\$ – cloakedlearning Jul 19 '16 at 13:06
  • \$\begingroup\$ thread_local exposed via API is a very bad idea for many reasons (for example because someone may cache it and C++ syntax won't stop it from happening). \$\endgroup\$ – bobah Jul 19 '16 at 14:01
  • \$\begingroup\$ right - I guess I didn't put enough thought into that. \$\endgroup\$ – cloakedlearning Jul 19 '16 at 15:39

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.