Epoll Multiplexed IO event

multiplex-epoll-server.cpp

// File: socket-echo-server.cpp
// Desc: Simple socket server with a single thread.

#include <iostream>
#include <string>
#include <vector>
#include <cassert>
#include <cstring> // memcpy

// ----- Unix/Linux headers -----//
#include <unistd.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <signal.h>

// Only available on Linux
#include <sys/epoll.h> 

struct Socket 
{
    int           sockfd; 
    sockaddr_in   sa;     
};

struct RecvMsg
{
    ssize_t     size;
    std::string msg;
};

void socket_close(Socket& sock);

void    fd_write(int fd, std::string const& text);
RecvMsg fd_read(int fd, size_t buffer_max = 500);

/** Make a server socket */
Socket socket_make_server(uint16_t port, const char* hostname = "0.0.0.0");
/** Accept connetion from a server socket and returns a client socket */
Socket socket_accept(Socket& server); 

/** Register file descriptor to be monitored by Epoll */
void Epoll_register_fd(int epoll_fd, int fd)
{
    epoll_event ev; 
    ev.events  = EPOLLIN;
    ev.data.fd = fd; 
    int res = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &ev);
    if(res == -1){
        throw std::runtime_error("Error: EPoll / unable to register file descritor");
    }    
}

/** Remove file descriptor from monitoring list. */
void Epoll_remove_fd(int epoll_fd, int fd)
{
    epoll_event ev;
    int res = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, &ev);
}

/*  Multiplex IO - Blocks current thread until any of monitored file descriptors 
 *  has 
 */
int Epoll_wait(int epoll_fd, size_t max_events, std::vector<epoll_event>& events)
{
    // Returns number of events received 
    int nfd = epoll_wait( epoll_fd     // Epoll file descripotr 
                        , &events[0]   // [output] Array containing received events 
                        , max_events   // Maximum number of events 
                        , -1           // Timeout (-1) => Block indefinitely
                        );

    return nfd;
}


int main(int argc, char **argv)
{
    std::puts(" [INFO] Program started. ");

    // --- Create socket file descriptor -----------------------//
    Socket server = socket_make_server(9026, "0.0.0.0");

    constexpr size_t MAX_EVENTS = 10;     
    int epfd = epoll_create1(0);
    assert(epfd != - 1);
    auto events = std::vector<epoll_event>(MAX_EVENTS);

    // Register stdin 
    Epoll_register_fd(epfd, STDIN_FILENO);

    // Register socket server 
    Epoll_register_fd(epfd, server.sockfd);

    // Contains client sockets file descriptors 
    std::vector<Socket> clients; 
    
 
    // Server-accept infinite loop 
    for(;;)
    {
        std::printf(" [TRACE] Epoll waiting for events \n");

        int nfsd = Epoll_wait(epfd, MAX_EVENTS, events);            
        assert( nfsd != - 1);

        std::printf(" [TRACE] Epoll events arrived => nfsd = %d \n", nfsd);

        for(int i = 0; i < nfsd; i++)
        {
            auto ev = events[i];
            
            std::printf(" ----------------------------------------------------\n");

            if( (ev.events & EPOLLERR) || (ev.events & EPOLLHUP) )
            {
                std::printf(" [ERROR] Epoll error has happened. \n");
                continue;
            }

            // STDIN event => It happens after user types something and
            // hit RETURN on terminal.
            if( ev.data.fd == STDIN_FILENO )
            {
                std::printf(" [TRACE] Received STDIN event \n");
                std::string line;
                std::getline(std::cin, line);
                std::cout << " \n STDIN = " + line << "\n";

                // Send typed message to all client sockets
                for(Socket& cli: clients)
                    fd_write(cli.sockfd, " STDIN = " + line + "\n");

                // Quit server 
                if(line == "exit"){ 
                    std::puts(" [INFO] Shutdown server OK.");
                    return EXIT_SUCCESS;
                }
                continue;
            }

            // Event that happens when there is a new connection.
            if(ev.data.fd == server.sockfd)
            {                
                Socket client = socket_accept(server);        
                Epoll_register_fd(epfd, client.sockfd);

                std::printf(" [TRACE] Accept client connection => ID = %d \n", client.sockfd);                
                fd_write(client.sockfd,  "\n => [SERVER INFO] Hello world client side!! \n");
                // socket_nonblock(client);
                clients.push_back(client);
                continue;
            }

            // Event that happens when some client socket sends data. 
            // EPOLIN event means that the file descriptor is ready for input. 
            if(ev.events & EPOLLIN)            
            {
               int fd = ev.data.fd;
               std::printf("\n [TRACE] Receive client event => ID = %d \n", fd);
               RecvMsg resp = fd_read(fd, 500);                                             
               if(resp.size == 0)
               {
                   std::printf("\n [CLIENT ID = %d] =>> Disconnected from server. \n", fd);
                   close(fd);
                   continue;
               }
               std::printf("\n [CLIENT ID = %d] =>> size = %d ; msg = %s \n", fd, resp.size, resp.msg.c_str());
               fd_write(fd, " [SERVER ECHO] => ID = " + std::to_string(fd) + " => " + resp.msg);
            }
            
        }


    } /* [[ ============== END client loop =========== ]] */

    socket_close(server);
    return 0;

} // --- End of main() -------------//

// ----------- Function Implementations ----------------------------------------//

void 
socket_close(Socket& sock)
{
    ::close(sock.sockfd);
    sock.sockfd = -1;    
}


void fd_write(int fd, std::string const& text)
{
    ::write(fd, text.c_str(), text.size());
}

RecvMsg fd_read(int fd, size_t buffer_max)
{
    std::string buffer(buffer_max, 0x00);
    ssize_t n = read(fd, &buffer[0], buffer_max);
    buffer.resize(n);
    return RecvMsg{n, buffer};
}


Socket 
socket_make_server(uint16_t port, const char* hostname)
{
    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
    // Returns (-1) on failure
    assert(sockfd != -1);

    // ---------- query address ------------------------------//
    // Note: the keyword 'struct' is not necessary here (redundant).
    struct hostent *h = gethostbyname(hostname);
    assert(h != nullptr);
    struct sockaddr_in sa;
    memcpy(&sa.sin_addr.s_addr, h->h_addr, h->h_length);
    sa.sin_family = AF_INET;
    // The function htons convert a number to big-endian format.
    sa.sin_port = htons(port);

    // ----- Bind to Port and wait for client connections ---------//
    if (::bind(sockfd, (sockaddr *)&sa, sizeof(sa)) == -1)            
        throw std::runtime_error("Error: unable to bind socket \n");
    

    // Enables binding to the same address
    int enable_reuseaddr = 1;
    if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &enable_reuseaddr, sizeof(int)) < 0)
        throw std::runtime_error("Error: unable to set socket option. \n");

    // Ignore SIGPIPE signal which would cause abnormal termination of socket server
    // Requires: #include <signal.h> header.
    signal(SIGPIPE, SIG_IGN);

    fprintf(stderr, " [TRACE] Listening client connection \n");
    int backlog = 5;
    assert(listen(sockfd, backlog) != -1);

    return Socket{ sockfd, sa };
}

Socket 
socket_accept(Socket& server)
{
    Socket client;
    socklen_t addrlen = sizeof(sockaddr_in);
    client.sockfd = accept(server.sockfd, (sockaddr *) &client.sa, &addrlen);
         
    if (client.sockfd == -1)
    {
        fprintf(stderr, " Error: failure to handle client socket. Check errno \n");
        close(client.sockfd);
    }
    return client;
}