//server.cpp

// CXXFLAGS = -std=c++2a

#include <sys/epoll.h>

#include <stdio.h>

#include <sys/socket.h>

#include <netinet/in.h>

#include <stdlib.h>

#include <string.h>

#include <fcntl.h>

#include <arpa/inet.h>

#include <unistd.h>

#include <errno.h>

#include <signal.h>

#include <thread>

#include <mutex>

#include <string>

#include <vector>

#define PRINT_LOG(fmt, ...) do \

{ \

? printf("%s:%d:%s: " fmt "\n", \

? ? ? __FILE__, __LINE__, __func__, ##__VA_ARGS__); \

} while (0) /* ; no trailing semicolon here */

#define PERROR(fmt, ...) do \

{ \

? printf("%s:%d:%s: %s. " fmt "\n", \

? ? ? __FILE__, __LINE__, __func__, strerror(errno), ##__VA_ARGS__); \

} while (0) /* ; no trailing semicolon here */

int efd;

std::vector<int> vfd;

std::mutex lock;

int time_to_write = 0;

int set_nonblock(int fd)

{

? int flags = fcntl(fd, F_GETFL);

? if (flags == -1) {

? ? PERROR("fcntl");

? ? return -1;

? }

? flags |= O_NONBLOCK;

? if (fcntl(fd, F_SETFL, flags) == -1) {

? ? PERROR("fcntl");

? ? return -1;

? }

? return 0;

}

void update_events()

{

? struct epoll_event ev;

? int ret = 0;

? int err = 0;

? for (;;){

? ? sleep(1); //test

? ? time_to_write = !time_to_write;

? ? std::lock_guard<std::mutex> guard(lock);

? ? // PRINT_LOG("vfd: %lu, time_to_write: %d", vfd.size(), time_to_write);

? ? for (auto fd : vfd){

? ? ? // PRINT_LOG("fd: %d, time_to_write: %d", fd, time_to_write);

? ? ? ev.events = EPOLLIN | EPOLLET;

? ? ? ev.data.fd = fd;

? ? ? if (time_to_write){

? ? ? ? ev.events = EPOLLOUT | EPOLLET;

? ? ? }

? ? ? ret = epoll_ctl(efd, EPOLL_CTL_MOD, fd, &ev);

? ? ? err = errno;

? ? ? if (err == EBADF){

? ? ? ? continue;

? ? ? }

? ? ? if (ret == -1) {

? ? ? ? PERROR("epoll_ctl");

? ? ? }

? ? }

? }

}

int main()

{

? struct epoll_event ev, events[10];

? int listenfd, connfd, nfds;

? /* set up listening socket */

? int port = 8000;

? int ret = 0;

? struct sockaddr_in my_addr, peer_addr;

? socklen_t addrlen = sizeof(peer_addr);

? int sockopt = 1;

? int seq = 0;

? memset(&my_addr, 0, sizeof(my_addr));

? my_addr.sin_port = htons(port);

? my_addr.sin_addr.s_addr = htons(INADDR_ANY);

? my_addr.sin_family = AF_INET;

? listenfd = socket(AF_INET, SOCK_STREAM, 0);

? if (listenfd == -1) {

? ? PERROR("socket");

? ? return -1;

? }

? ret = setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &sockopt,

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? sizeof(sockopt));

? if (ret == -1){

? ? PERROR("setsockopt");

? ? close(listenfd);

? ? return -1;

? }

? ret = bind(listenfd, (struct sockaddr*)&my_addr, sizeof(my_addr));

? if (ret == -1) {

? ? PERROR("bind");

? ? close(listenfd);

? ? return -1;

? }

? ret = listen(listenfd, SOMAXCONN);

? if (ret == -1) {

? ? PERROR("listen");

? ? close(listenfd);

? ? return -1;

? }

? PRINT_LOG("listen on port: %d", port);

? signal(SIGPIPE, SIG_IGN);

? /* set up epoll */

? efd = epoll_create1(0);

? if (efd == -1) {

? ? PERROR("epoll_create1");

? ? return -1;

? }

? ev.events = EPOLLIN;

? ev.data.fd = listenfd;

? if (epoll_ctl(efd, EPOLL_CTL_ADD, listenfd, &ev) == -1) {

? ? PERROR("epoll_ctl");

? ? return -1;

? }

? PRINT_LOG("waiting for data input from clients");

? std::thread(update_events).detach();

? for (;;) {

? ? nfds = epoll_wait(efd, events, sizeof(events) / sizeof(events[0]),

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? 1000); //1 sec

? ? // PRINT_LOG("epoll_wait");

? ? if (nfds == -1) {

? ? ? PERROR("epoll_wait");

? ? ? for (int i = 0; i != sizeof(events) / sizeof(events[0]); i++){

? ? ? ? PRINT_LOG();

? ? ? ? close(events[i].data.fd);

? ? ? }

? ? ? return -1;

? ? }

? ? for (int i = 0; i < nfds; i++) {

? ? ? if (events[i].data.fd == listenfd) {

? ? ? ? connfd = accept(listenfd,

? ? ? ? ? ? ? ? (struct sockaddr *) &peer_addr, &addrlen);

? ? ? ? if (connfd == -1) {

? ? ? ? ? PERROR("accept");

? ? ? ? ? continue;

? ? ? ? }

? ? ? ? if (set_nonblock(connfd)){

? ? ? ? ? PRINT_LOG("set_nonblock failed");

? ? ? ? ? continue;

? ? ? ? }

? ? ? ? PRINT_LOG("accept client %s:%u", inet_ntoa(peer_addr.sin_addr),

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? peer_addr.sin_port);

? ? ? ? ev.events = EPOLLIN | EPOLLET;

? ? ? ? ev.data.fd = connfd;

? ? ? ? if (epoll_ctl(efd, EPOLL_CTL_ADD, connfd, &ev) == -1) {

? ? ? ? ? PERROR("epoll_ctl");

? ? ? ? }

? ? ? ? {

? ? ? ? ? std::lock_guard<std::mutex> guard(lock);

? ? ? ? ? vfd.push_back(connfd);

? ? ? ? }

? ? ? } else if (events[i].events & (EPOLLHUP | EPOLLERR)) {

? ? ? ? int fd = events[i].data.fd;

? ? ? ? close(fd);

? ? ? ? {

? ? ? ? ? std::lock_guard<std::mutex> guard(lock);

? ? ? ? ? std::erase(vfd, fd); // -std=c++2a

? ? ? ? }

? ? ? ? PRINT_LOG("EPOLLHUP | EPOLLERR");

? ? ? } else if (events[i].events & EPOLLIN) {

? ? ? ? int fd = events[i].data.fd;

? ? ? ? enum {count = 1024};

? ? ? ? char buf[count + 1] = {'\0'}, *p = buf;

? ? ? ? size_t len = 0;

? ? ? ? while (len < count){

? ? ? ? ? ret = read(fd, p, count - len);

? ? ? ? ? int err = errno;

? ? ? ? ? if (ret > 0){

? ? ? ? ? ? len += ret;

? ? ? ? ? ? p += ret;

? ? ? ? ? }

? ? ? ? ? if (ret == 0){

? ? ? ? ? ? close(fd);

? ? ? ? ? ? {

? ? ? ? ? ? ? std::lock_guard<std::mutex> guard(lock);

? ? ? ? ? ? ? std::erase(vfd, fd);

? ? ? ? ? ? }

? ? ? ? ? ? break;

? ? ? ? ? }

? ? ? ? ? if (err == EAGAIN || err == EWOULDBLOCK || err == EPIPE ||

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? err == ECONNRESET){

? ? ? ? ? ? break;

? ? ? ? ? }

? ? ? ? ? if (err == EINTR){

? ? ? ? ? ? continue;

? ? ? ? ? }

? ? ? ? ? if (err != 0){

? ? ? ? ? ? PRINT_LOG("ret: %d, err: %d, %s", ret, err,

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? strerror(err));

? ? ? ? ? }

? ? ? ? }

? ? ? ? if (len > 0) {

? ? ? ? ? PRINT_LOG("%s", buf);

? ? ? ? }

? ? ? } else if (events[i].events & EPOLLOUT){

? ? ? ? int fd = events[i].data.fd;

? ? ? ? enum {count = 1024};

? ? ? ? char buf[count + 1] = {'\0'}, *p = buf;

? ? ? ? std::string msg = "hello client " + std::to_string(seq++);

? ? ? ? strncpy(buf, msg.c_str(), count);

? ? ? ? size_t len = 0;

? ? ? ? while (len < count){

? ? ? ? ? ret = write(fd, p, count - len);

? ? ? ? ? int err = errno;

? ? ? ? ? if (ret > 0){

? ? ? ? ? ? len += ret;

? ? ? ? ? ? p += ret;

? ? ? ? ? }

? ? ? ? ? if (ret == 0){

? ? ? ? ? ? close(fd);

? ? ? ? ? ? {

? ? ? ? ? ? ? std::lock_guard<std::mutex> guard(lock);

? ? ? ? ? ? ? std::erase(vfd, fd);

? ? ? ? ? ? }

? ? ? ? ? ? break;

? ? ? ? ? }

? ? ? ? ? if (err == EAGAIN || err == EWOULDBLOCK || err == EPIPE ||

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? err == ECONNRESET){

? ? ? ? ? ? break;

? ? ? ? ? }

? ? ? ? ? if (err == EINTR){

? ? ? ? ? ? continue;

? ? ? ? ? }

? ? ? ? ? if (err != 0){

? ? ? ? ? ? PRINT_LOG("ret: %d, err: %d, %s", ret, err,

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? strerror(err));

? ? ? ? ? }

? ? ? ? }

? ? ? } else {

? ? ? ? PRINT_LOG("unknown event");

? ? ? ? int fd = events[i].data.fd;

? ? ? ? close(fd);

? ? ? ? {

? ? ? ? ? std::lock_guard<std::mutex> guard(lock);

? ? ? ? ? std::erase(vfd, fd);

? ? ? ? }

? ? ? }

? ? }

? }

? return 0;

}

---

//client.cpp

#include <sys/socket.h>

#include <unistd.h>

#include <fcntl.h>

#include <stdio.h>

#include <string.h>

#include <errno.h>

#include <stdlib.h>

#include <arpa/inet.h>

#include <signal.h>

#include <string>

#define PRINT_LOG(fmt, ...) do \

{ \

? printf("%s:%d:%s: " fmt "\n", \

? ? ? __FILE__, __LINE__, __func__, ##__VA_ARGS__); \

} while (0) /* ; no trailing semicolon here */

#define PERROR(fmt, ...) do \

{ \

? printf("%s:%d:%s: %s. " fmt "\n", \

? ? ? __FILE__, __LINE__, __func__, strerror(errno), ##__VA_ARGS__); \

} while (0) /* ; no trailing semicolon here */

int set_nonblock(int fd)

{

? int flags = fcntl(fd, F_GETFL);

? if (flags == -1) {

? ? PERROR("fcntl");

? ? return -1;

? }

? flags |= O_NONBLOCK;

? if (fcntl(fd, F_SETFL, flags) == -1) {

? ? PERROR("fcntl");

? ? return -1;

? }

? return 0;

}

int main(int argc, char *argv[])

{

? int fd;

? int ret;

? struct sockaddr_in addr;

? const char *host;

? int port;

? int seq = 0;

? if (argc != 4) {

? ? PRINT_LOG("Usage: %s <host> <port> <msg>\n", argv[0]);

? ? return 1;

? }

? host = argv[1];

? port = atoi(argv[2]);

? fd = socket(AF_INET, SOCK_STREAM, 0);

? if (fd == -1){

? ? PERROR("socket");

? ? return -1;

? }

? memset(&addr, 0, sizeof(addr));

? addr.sin_port = htons(port);

? addr.sin_family = AF_INET;

? ret = inet_aton(host, (struct in_addr*)&addr.sin_addr.s_addr);

? if (ret == 0){

? ? PRINT_LOG("inet_aton failed");

? ? return -1;

? }

? ret = connect(fd, (struct sockaddr*)&addr, sizeof(addr));

? if (ret == -1){

? ? PERROR("connect");

? ? close(fd);

? ? return -1;

? }

? PRINT_LOG("connect to server %s:%d", host, port);

? set_nonblock(fd);

? // signal(SIGPIPE, SIG_IGN);

? for(;;){

? ? //write

? ? sleep(1); //test

? ? std::string msg = argv[3] + std::to_string(seq++);

? ? ret = write(fd, msg.c_str(), msg.size()); //

? ? //read

? ? enum {count = 1024};

? ? char buf[count + 1] = {'\0'};

? ? ret = read(fd, buf, count); //

? ? if (ret > 0){

? ? ? PRINT_LOG("%s", buf);

? ? }

? }

? ret = close(fd);

? return 0;

}