0
\$\begingroup\$

I created an asio framework using epoll.

Full project: https://github.com/arkceajin/EpollSocket.git

I wanna know is there any potential issue.

Below is the core part, and the remaining part is the application and serialization code.

The abstraction of socket connection.

abstractsocket.h

#ifndef ABSTRACTSOCKET_H
#define ABSTRACTSOCKET_H

#include <string>
#include <sys/socket.h>
#include <netdb.h>
#include <vector>
#include <cstddef>
#include "pack.h"

#define UNUSED(x) (void)x;
#define EPOLL_QUEUE_LEN          100
#define MAX_EPOLL_EVENTS_PER_RUN 100
#define EPOLL_WAIT_TIMEOUT       500

namespace EpollSocket {

typedef uint8_t Byte;
typedef std::vector<Byte> Bytes;
typedef std::vector<int> Clients;

typedef void (*OnNewConnected)(int fd, const char* address);
typedef void (*OnServerReceived)(int fd, const Clients& clients);
typedef void (*OnClientReceived)(int fd);

void receive(const int& fd, Bytes& buffer, const size_t& size);
void send(const int& fd, const Bytes &buffer);

class SocketException : public std::exception
{
public:
    enum Type {
        SocketBindError,
        SocketReceiveError,
        SocketSendError,
        SocketUnknowError,
    };

    SocketException(const Type& type);
    ~SocketException() override;

    int perrno() const;
private:
    const char* what() const noexcept override;
    Type mType;
};

class AbstractSocket
{
public:
    enum SocketState{
        UnconnectedState,
        HostLookupState,
        ConnectingState,
        ConnectedState,
        BoundState,
        ClosingState,
        ListeningState
    };

    AbstractSocket();

    bool close();

    inline int sockfd() const {
      return mSockfd;
    }

protected:
    int setsockopt(const int& optname, const int &val);

    bool establish(const std::string &addr,
                   const std::string &service,
                   int (*action)(int __fd, __CONST_SOCKADDR_ARG __addr, socklen_t __len),
                   const int& socktype);



    int         mSockfd;
    SocketState mState;
};

}

abstractsocket.cpp

#endif // ABSTRACTSOCKET_H
#include "abstractsocket.h"
#include <arpa/inet.h>
#include <sys/un.h>
#include <unistd.h>
#include <sys/epoll.h>

namespace EpollSocket {

SocketException::SocketException(const SocketException::Type &type):
    std::exception(),
    mType(type)
{

}

SocketException::~SocketException()
{

}

int SocketException::perrno() const
{
    return errno;
}

const char *SocketException::what() const noexcept
{
    switch (mType) {
        case SocketBindError:
            return "Failed to bind";
        case SocketReceiveError:
            return "SocketReceiveError";
        case SocketSendError:
            return "SocketSendError";
        case SocketUnknowError:
            return "SocketUnknowError";
    }
    return "SocketUnknowError";
}

AbstractSocket::AbstractSocket():
    mSockfd(-1),
    mState(UnconnectedState)
{

}

int AbstractSocket::setsockopt(const int &optname, const int& val)
{
    return ::setsockopt(mSockfd, SOL_SOCKET, optname, &val, sizeof val);
}

bool AbstractSocket::establish(const std::string &addr,
                               const std::string &service,
                               int (*action)(int __fd, __CONST_SOCKADDR_ARG __addr, socklen_t __len),
                               const int &socktype)
{
    if(mState != UnconnectedState)
        return false;
    int status;
    addrinfo hints;
    addrinfo* servinfo = nullptr, *p = nullptr;

    memset(&hints, 0, sizeof hints);
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = socktype;
    if(addr.empty())
        hints.ai_flags = AI_PASSIVE;

    if((status = getaddrinfo(addr.empty()?nullptr:addr.c_str(),
                             service.c_str(),
                             &hints, &servinfo)) != 0) {
        fprintf(stderr, "getaddrinfo error: %s\n", gai_strerror(status));
        return false;
    }

    // loop through all the results and bind to the first we can
    for(p = servinfo; p != nullptr; p = p->ai_next) {
        if((mSockfd = socket(p->ai_family, p->ai_socktype, p->ai_protocol)) == -1) {
            perror("socket error"); // Not a critical error.
            continue;
        }
        if(action(mSockfd, p->ai_addr, p->ai_addrlen) == -1)
            continue;
        break;
    }

    freeaddrinfo(servinfo);

    if (p == nullptr)  {
        throw SocketException(SocketException::SocketBindError);
    }
    mState = ConnectingState;
    return true;
}

bool AbstractSocket::close()
{
    const int& r = ::close(mSockfd);
    if (r != -1) {
        mState = UnconnectedState;
        return true;
    }
    return false;
}

void receive(const int &fd, Bytes &buffer, const size_t &size)
{
    if (size == 0)
        return;
    size_t bytesRead = 0;
    ssize_t result;
    while (bytesRead < size) {
        result = ::read(fd, static_cast<Byte*>(buffer.data()) + bytesRead, size - bytesRead);
        if (result < 0 ) {
            throw SocketException(SocketException::SocketReceiveError);
        } else if (result == 0) {
            break;
        }
        bytesRead += static_cast<size_t>(result);
    }
}

void send(const int &fd, const Bytes &buffer)
{
    size_t bytesSend = 0;
    ssize_t result;
    while (bytesSend < buffer.size()) {
        result = ::write(fd, const_cast<Byte*>(buffer.data()) + bytesSend, buffer.size());
        if (result < 0 ) {
            throw SocketException(SocketException::SocketSendError);
        }
        bytesSend += static_cast<size_t>(result);
    }
}

} //EpollSocket

The server side

server.h

#ifndef SERVER_H
#define SERVER_H

#include "abstractsocket.h"

namespace EpollSocket {

class Server : public AbstractSocket
{
public:
    Server(const int& backlog = SOMAXCONN);

    bool listen(const std::string& service,
                OnNewConnected onNewConnected,
                OnServerReceived onReceived,
                const std::string& addr = std::string(),
                const int &socktype = SOCK_STREAM);
private:
    /**
     * @brief backlog is the number of connections allowed on the incoming queue.
     */
    int mBacklog;

    Clients mClients;

};

}


#endif // SERVER_H

server.cpp

#include "server.h"
#include <sys/socket.h>
#include <stdio.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <fcntl.h>
#include <sys/un.h>
#include <netdb.h>
#include <algorithm>

namespace EpollSocket {

// get sockaddr, IPv4 or IPv6:
void *get_in_addr(struct sockaddr *sa)
{
    if (sa->sa_family == AF_INET) {
        return &((reinterpret_cast<struct sockaddr_in*>(sa))->sin_addr);
    }
    return &((reinterpret_cast<struct sockaddr_in6*>(sa))->sin6_addr);
}

bool set_nonblocking(const int& sfd)
{
    int flags = fcntl(sfd, F_GETFL);
    if(flags < 0) {
        perror("fcntl F_GETFL");
        return false;
    }
    flags |= O_NONBLOCK;
    if(fcntl(sfd, F_SETFL, flags) < 0) {
        perror("fcntl F_SETFL");
        return false;
    }
    return true;
}

Server::Server(const int &backlog):
    AbstractSocket(),
    mBacklog(backlog)
{

}

bool Server::listen(const std::string &service,
                    OnNewConnected onNewConnected,
                    OnServerReceived onReceived,
                    const std::string &addr,
                    const int &socktype)
{
    if(!AbstractSocket::establish(addr, service, ::bind, socktype)){
        AbstractSocket::close();
        return false;
    }
#if 0 // Don't use this in product enviroment
    if(AbstractSocket::setsockopt(SO_REUSEADDR, 1) == -1){
        perror("setsockopt SO_REUSEADDR error");
        AbstractSocket::close();
        return false;
    }
#endif

    if(::listen(mSockfd, mBacklog) == -1) {
        perror("listen error");
        AbstractSocket::close();
        return false;
    }

    AbstractSocket::mState = AbstractSocket::ConnectedState;

    if(!set_nonblocking(mSockfd)) {
        AbstractSocket::close();
        return false;
    }

    const int& epfd = epoll_create(EPOLL_QUEUE_LEN);

    struct epoll_event ev;
    //ev.events = EPOLLIN | EPOLLET;
    ev.events = EPOLLIN;
    ev.data.fd = mSockfd;

    if(epoll_ctl(epfd, EPOLL_CTL_ADD, mSockfd, &ev) == -1) {
        perror("epoll_ctl error");
        AbstractSocket::close();
        return false;
    }

    struct epoll_event events[MAX_EPOLL_EVENTS_PER_RUN];
    char remoteIP[INET6_ADDRSTRLEN];
    struct sockaddr_storage remoteaddr; // client address
    socklen_t addrlen;

    while (AbstractSocket::mState == AbstractSocket::ConnectedState) {
        int nfds = epoll_wait(epfd, events, MAX_EPOLL_EVENTS_PER_RUN, EPOLL_WAIT_TIMEOUT);
        if(nfds == -1) {
            perror("epoll_wait error");
            AbstractSocket::close();
            return false;
        }
        for(int i = 0; i < nfds; i++) {
            const epoll_event& e = events[i];
            if(e.data.fd == mSockfd) {// newly accept()ed socket descriptor
                addrlen = sizeof remoteaddr;
                const int& newfd = accept(mSockfd, reinterpret_cast<struct sockaddr*>(&remoteaddr), &addrlen);
                if(newfd == -1) {
                    perror("accept error");
                    continue;
                }
                ev.data.fd = newfd;
                ev.events = EPOLLIN | EPOLLET;
                if(!set_nonblocking(newfd)) {
                    perror("set O_NONBLOCK error");
                    ::close(newfd);
                    continue;
                }

                if(epoll_ctl(epfd, EPOLL_CTL_ADD, newfd, &ev) == -1) {
                    perror("epoll_ctl error");
                    ::close(newfd);
                }
                mClients.emplace_back(newfd);
                onNewConnected(newfd,
                               inet_ntop(remoteaddr.ss_family, get_in_addr(reinterpret_cast<struct sockaddr*>(&remoteaddr)), remoteIP, INET6_ADDRSTRLEN));
            } else if(e.events & EPOLLIN) {
                const int& fd = e.data.fd;
                if(fd < 0) {
                    perror("EPOLLIN fd error");
                    continue;
                }
                printf("EPOLLIN receive\n");
                // start read
                try {
                    onReceived(fd, mClients);
                } catch (const SocketException&) {
                    //disconnected
                    printf("start disconnected: %d\n", errno);
                    epoll_ctl(epfd, EPOLL_CTL_DEL, fd, &ev);
                    ::close(fd);
                    mClients.erase(std::remove(mClients.begin(), mClients.end(), fd), mClients.end());
                    printf("end disconnected: %d\n", errno);
                }
            } else if(e.events & EPOLLOUT) {
                // start write?
            }
        }
    }
    return true;
}

} // EpollSocket

The client side

#ifndef CLIENT_H
#define CLIENT_H

#include "abstractsocket.h"

namespace EpollSocket {

class Client : public AbstractSocket
{
public:
    Client();

    bool connect(const std::string& addr,
                 const std::string& service,
                 const OnClientReceived onReceived,
                 const int &socktype = SOCK_STREAM);

private:

};

}

#endif // CLIENT_H

client.cpp

#include "client.h"
#include <sys/epoll.h>
#include <cstring>
#include <iostream>

namespace EpollSocket {

Client::Client() :
    AbstractSocket()
{

}

bool Client::connect(const std::string &addr,
                     const std::string &service,
                     const OnClientReceived onReceived,
                     const int &socktype)
{
    if(!AbstractSocket::establish(addr, service, ::connect, socktype)){
        AbstractSocket::close();
        return false;
    }

    AbstractSocket::mState = AbstractSocket::ConnectedState;

    const int& epfd = epoll_create(EPOLL_QUEUE_LEN);

    struct epoll_event ev;
    ev.events = EPOLLIN;
    ev.data.fd = mSockfd;

    if(epoll_ctl(epfd, EPOLL_CTL_ADD, mSockfd, &ev) == -1) {
        perror("epoll_ctl error");
        AbstractSocket::close();
        return false;
    }

    struct epoll_event events[MAX_EPOLL_EVENTS_PER_RUN];

    while (AbstractSocket::mState == AbstractSocket::ConnectedState) {
        int nfds = epoll_wait(epfd, events, MAX_EPOLL_EVENTS_PER_RUN, EPOLL_WAIT_TIMEOUT);
        if(nfds == -1) {
            perror("epoll_wait error");
            AbstractSocket::close();
            return false;
        }
        for(int i = 0; i < nfds; i++) {
            const epoll_event& e = events[i];
            if(e.events & EPOLLIN) {
                const int& fd = e.data.fd;
                if(fd < 0) {
                    perror("EPOLLIN fd error");
                    continue;
                }
                // start read
                try {
                    onReceived(mSockfd);
                } catch (const SocketException&) {
                    //disconnected
                    printf("start disconnected: %d\n", errno);
                    epoll_ctl(epfd, EPOLL_CTL_DEL, mSockfd, &ev);
                    AbstractSocket::close();
                    printf("end disconnected: %d\n", errno);
                }
            }
        }
    }
    return true;
}


} // EpollSocket
\$\endgroup\$
  • \$\begingroup\$ Dont know if you are aware, but boost.asio does what you are trying to do without reinvent the wheel. \$\endgroup\$ – camp0 Feb 3 at 10:27
  • \$\begingroup\$ There is nothing related to epoll() in the code you posted. What exactly do you want reviewed? Make sure that the things you actually want reviewed are included fully in your question, don't just give a link to it. \$\endgroup\$ – G. Sliepen Feb 3 at 10:50
  • \$\begingroup\$ @G.Sliepen Hi, thanks for the reply, I add the core part of the code which I want to be reviewed. \$\endgroup\$ – JustWe Feb 3 at 11:29
1
\$\begingroup\$

Prefer static const variables over #defines

Don't #define constants if you could just as well declare them as regular static const variables. One advantage of this is that you can put those variables inside a namespace, whereas macros are always globally visible.

Avoid giving new names to existing types

Don't make a new type Byte when it's just the same as uint8_t. It saves hardly any typing, and now someone reading your code first has to figure out what a Byte actually is, whereas uint8_t is a standard type that should be known to anyone.

If you can use C++17, then you can use std::byte.

Apart from Bytes, the other typedefs in abstractsocket.h are not used at all in either abstractsocket.h or abstractsocket.cpp. So they should not be there at all.

Make send() and receive() member functions of AbstractSocket

These functions are supposed to be used on sockets, so make this explicit: they should be member functions of AbstractSocket.

Inherit from std::runtime_error instead of std::exception

Instead of making SocketException inherit from std::exception, make it inherit std::runtime_error. The latter is more specific, so it provides more information to the application.

Use further inheritance for exceptions

Instead of having enum Type to distinguish between possible socket exceptions, just use more inheritance to follow the pattern already used by exceptions. So:

class SocketException: public std::runtime_error {...};
class SocketBindError: public SocketException {...};
class SocketReceiveError: public SocketException {...};
class SocketSendError: public SocketException {...};

Avoid repeating yourself

There is some repetition going on in your class names. For example, SocketExpection is part of namespace EpollSocket, so "Socket" appears twice. You could get rid of some duplication there.

AbstractSocket is not abstract at all

Your class AbstractSocket is actually a concrete socket implementation, so why does it have "Abstract" in the name?

Be consistent in reporting errors

Your code is printing errors in many different ways:

  • fprintf(stderr, ...)
  • printf(...)
  • perror(...)

Make sure there is a consistent way of printing errors, if this is at all desired. If this is meant to be a library, it's probably best not to print anything, and let the application that calls these library functions decide how to report errors.

Also, sometimes you return false, other times you throw a SocketException. While there are sometimes good reasons to have both ways of returning an error, there doesn't seem to be much consistency in your code. In general, exceptions should be used for really exceptional conditions. Network errors however are quite common. I would recommend you use the boolean return type to indicate whether a function succeeded or not, and possibly have the functions take a reference to a std::error_code so it can fill in more details about the error in it.

You are not binding outgoing sockets

In AbstractSocket::establish(), you are calling socket() to create an outgoing socket, but don't call bind() on it. That's is perfectly normal. But when it fails, you are throwing a SocketBindError. That is incorrect, it is more likely it's a connection error, so you should add a SocketConnectError.

Do cleanup on errors in establish(), not in the action() callback

It's always best to clean up resources in the same scope as where you created them. So if a connection fails, then call close() in establish().

Avoid capturing return values by reference

In AbstractSocket::close() you write this line:

const int &r = ::close(mSockfd);

That's quite weird. By using a reference here, the return value of ::close() is actually stored in a temporary, and you are creating a const reference to a temporary value. Luckily, in C++11 and later, this extends the lifetime of the temporary until the end of the function scope, so it is not undefined behavior, but the normal way to write this is:

int r = ::close(mSockfd);

A call to ::close() always closes the socket, even if it returns an error

So in AbstractSocket::close(), you should unconditionally set mState = UnconnectedState.

receive() and send() are broken

You are calling receive() with a given size, but there is no guarantee that that many bytes have actually been received by the kernel. In fact, you have no guarantee that any bytes have been received; epoll() might spuriously returned and set the POLLIN flag even if nothing is received. In any case, if you call receive(..., 1000), and only 900 bytes were received so far by the kernel, then the first call to ::read() will succeed and return 900, but the second one will return -1 and set errno to EWOULDBLOCK.

Similarly, send() will also fail if the size of buffer if the other side doesn't process the data fast enough and the sending side's kernel buffers are filled up.

These things are normal behavior, you have to deal with that in your code instead of throwing an exception.

Use getnameinfo() instead of inet_ntop()

The counterpart to getaddrinfo() is getnameinfo(). It takes a struct sockaddr *, so you don't have to do the trick with get_in_addr() to give inet_ntop() a pointer to the actual address. While getnameinfo() normally tries to resolve an address to a hostname, if you only want to see a numeric address you can pass NI_NUMERICHOST | NI_NUMERICSERV as the flags argument.

You are not implementing asynchronous I/O at all

There is nothing asynchronous about your code. It is just a basic event loop, where you sequentially process events as they are coming in. A real asynchronous I/O framework would allow the application to submit multiple concurrent read and write requests, and have the framework process this in the background while the application can do something else.

| improve this answer | |
\$\endgroup\$

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.