C++ Socket syncronous poll wrapper

Question

Below is a simple C++ abstraction over linux sockets, and a layer of abstraction over a synchronous poll call. The goal was to exploit move semantics to allow the Sockets to be used as value types.

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};
}

Socketset.h

#pragma once
#include <poll.h>
#include <map>
#include <vector>
#include <memory>
#include "socket.h"

//Collection of sockets, mutliplex over them with poll

class SocketSet {

public:
    SocketSet();
    SocketSet(const SocketSet&) = delete;
    SocketSet& operator=(const SocketSet&) = delete;

    void add_socket(std::shared_ptr<Socket>);
    void remove_socket(std::shared_ptr<Socket>);
    bool poll(int timeout_ms);

    std::shared_ptr<Socket> get_next();

private:
    using Pollfd = struct pollfd;

    std::map<int, std::shared_ptr<Socket>> sockets;
    std::vector<Pollfd> pfds;
};

socketset.cpp

#include <sys/types.h>
#include <sys/socket.h>
#include <stdexcept>
#include <cstring>
#include <iostream>
#include <algorithm>

#include "socketset.h"

SocketSet::SocketSet(){}

void SocketSet::add_socket(std::shared_ptr<Socket> socket){
    sockets[socket->getsd()] = socket;
}

void SocketSet::remove_socket(std::shared_ptr<Socket> socket){
    for(auto it = sockets.begin(); it != sockets.end();){
        if (it->second->getsd() == socket->getsd()){
            it = sockets.erase(it);
        } else {
            ++it;
        }
    }
}

bool SocketSet::poll(int timeout_ms){
    pfds.clear();
    for(const auto &kv : sockets){
        pfds.push_back({kv.first, POLLIN, 0});
    }
    int ret = ::poll(&pfds[0], pfds.size(), timeout_ms);
    if(ret < 0){
        throw std::runtime_error(std::string("poll: ") + std::strerror(errno));
    }

    return !!ret;
}

std::shared_ptr<Socket> SocketSet::get_next(){
    for(auto &p : pfds){
        if(p.revents & (POLLHUP | POLLERR | POLLNVAL)){
            sockets[p.fd]->has_error = true;
            return sockets[p.fd];
        }

        if(p.revents & POLLIN){
            p.revents = 0;
            return sockets[p.fd];
        }
    }

    //signal that we are done with nullptr
    return nullptr;
}

A simple client of this API might look something like:

main.cpp

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

int main(){
    std::vector<std::shared_ptr<Socket>> sockets;
    auto listen_socket = std::make_shared<ListenSocket>(33333);
    SocketSet ss;
    ss.add_socket(std::static_pointer_cast<Socket>(listen_socket));

    while (true){
        if (ss.poll(500)){
            while (auto sock = ss.get_next()){
                try {
                    if(sock->has_error) {
                        throw std::runtime_error("bad poll");
                    }

                    if(sock == listen_socket){
                        auto ds = std::make_shared<DataSocket>(listen_socket->accept());
                        ss.add_socket(ds);
                        sockets.push_back(std::static_pointer_cast<Socket>(ds));
                    } else {
                        auto s = std::static_pointer_cast<DataSocket>(sock);
                        char buf[5];
                        s->recv(buf, 4);
                        buf[5] = '\0';
                        std::cout << buf << std::endl;
                    }
                } catch (const std::runtime_error &e){
                    ss.remove_socket(sock);
                    sockets.erase(std::remove_if(sockets.begin(), sockets.end(), [sock](std::shared_ptr<Socket> s){
                        return s == sock;
                    }));
                }
            }
        }
    }
}

Note that the fixed size recv call is sufficient for my use case since I intend to use these classes with binary encoded data (however in makes my example code look a bit wonky).

I'm most interested in a review of the socketset abstraction since I'm not convinced I have the correct design. The shared pointers are quite heavy weight and are forced onto the user of the API. I tried to do something with unique pointers and raw pointers but it becomes very easy to accidentally invalidate them. Use value types makes it hard to have polymorphic sockets due to slicing.

Answer 1

Your socket abstraction doesn't offer many features. For example, I don't see any way of closing a TCP connection. get_peer_name() could be made available on any type of socket and you could also expose IP addresses with proper data structures instead of the simply strings.

The names DataSocket and ConnectSocket don't really tell me if they are TCP or UDP sockets. By looking further at the code I notice that they are for TCP connections. Why do you need such a hierarchy? Currently, if one constructs a DataSocket what can it be used for? It doesn't have any connection-related methods available (those are only present in subclasses).

The SocketSet appears difficult to use and not following normal C++ idioms. It exposes too many details on how it's operating, leaving you to continually add and remove sockets from it (while at the same time also adding/removing them from your vector). As you already declare a vector of sockets, I would see the API proving a PollService which (periodically) takes an iterator pair of sockets as input and offers an iterator for iterating over the sockets on which one can act.

std::vector<std::shared_ptr<Socket>> sockets;
PollService pollService(500);

while (true)
{
   for (auto& socket : pollService.poll(begin(sockets), end(sockets)))
   {
      //do something with socket
   }
}

Answer 2

Leaky abstraction

Your Socket contains a public method int getsd() which returns a raw handle to the socket. With it, a client can call any methods it wants against the socket (send/close/recv) without having to use your abstraction layer that you've built. This method shouldn't exist, it undermines all of the other protections you've put in place. This can obviously cause issues, since for example your poll call in SocketSet needs access to the raw handle, however the number of classes that need access to the raw handle should be limited and IMO explicitly managed through the use of friends.

Naming

DataSocket is a bit too close to datagram socket, which isn't what it's modelling, it's constructor explicitly sets it up as a stream socket. I prefer to model it as such by either calling it StreamSocket or ConnectedSocket.

Connect is an action, so is something I would expect to be used in a method name, not in a class name. Connectable would be better (although personally I would merge the functionality between ConnectSocket and DataSocket together because they both only really work on streamed sockets.

ConnectSocket - DeadEnd

ConnectSocket provides a default constructor which doesn't do anything (ConnectSocket(){}). Why?

Protected members

ConnectSocket stores information as protected members (port and addr). addr doesn't appear to ever be used and port is only ever used in the constructor, why isn't it just a local variable? There's a similar situation with ListenSocket, port is never set, there is a constant backlog that is protected (does it need to be, rather than private? does it even need to be a class level constant rather than a method level?)

remove_socket

It looks to me like remove_socket iterates through all of the sockets, even after it has found the one it's looking for. Are you expecting the socket to be in the list more than once, or should it return after removing the first one?

void SocketSet::remove_socket(std::shared_ptr<Socket> socket) {
    for (auto it = sockets.begin(); it != sockets.end();) {
        if (it->second->getsd() == socket->getsd()) {
            it = sockets.erase(it);
            // do we need to keep going here?
        }
        else {
            ++it;
        }
    }
}