Q:

C++ Socket syncronous poll wrapper

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.

A:

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

