Commit 61bd6331 authored by Mathieu Stefani's avatar Mathieu Stefani Committed by Mathieu STEFANI

Initial implementation of a multi-threaded TCP listener

parent ca15b634
cmake_minimum_required (VERSION 2.8.7)
project (restgear)
if(CMAKE_COMPILER_IS_GNUCXX)
add_definitions(-std=c++0x)
endif()
add_subdirectory (src)
include_directories (src)
add_executable(server main.cc)
target_link_libraries(server net)
#include "listener.h"
#include "net.h"
#include <iostream>
#include <cstring>
using namespace std;
int main() {
Net::Address addr(Net::Ipv4::any(), 9080);
Net::Tcp::Listener listener(addr);
if (listener.bind()) {
cout << "Now listening on " << addr.host() << " " << addr.port() << endl;
listener.run();
}
else {
cout << "Failed to listen lol" << endl;
cout << "errno = " << strerror(errno) << endl;
}
}
set(SOURCE_FILES
listener.cc
net.cc
peer.cc
)
add_library(net ${SOURCE_FILES})
include_directories (${CMAKE_CURRENT_SOURCE_DIR})
target_link_libraries(net pthread)
/* common.h
Mathieu Stefani, 12 August 2015
A collection of macro / utilities / constants
*/
#pragma once
#include <cstdio>
#include <cassert>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#define TRY(...) \
do { \
auto ret = __VA_ARGS__; \
if (ret < 0) { \
perror(#__VA_ARGS__); \
cerr << gai_strerror(ret) << endl; \
return false; \
} \
} while (0)
#define TRY_RET(...) \
[&]() { \
auto ret = __VA_ARGS__; \
if (ret < 0) { \
perror(#__VA_ARGS__); \
throw std::runtime_error(#__VA_ARGS__); \
} \
return ret; \
}(); \
(void) 0
namespace Const {
static constexpr int MaxBacklog = 128;
static constexpr int MaxEvents = 1024;
static constexpr int MaxBuffer = 4096;
static constexpr int ChunkSize = 1024;
}
/* listener.cc
Mathieu Stefani, 12 August 2015
*/
#include <thread>
#include <iostream>
#include <sys/socket.h>
#include <unistd.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <sys/epoll.h>
#include <cassert>
#include <cstring>
#include "listener.h"
#include "peer.h"
#include "common.h"
using namespace std;
namespace Net {
namespace Tcp {
IoWorker::IoWorker() {
epoll_fd = TRY_RET(epoll_create(128));
}
IoWorker::~IoWorker() {
if (thread && thread->joinable()) {
thread->join();
}
}
void
IoWorker::start() {
thread.reset(new std::thread([this]() {
this->run();
}));
}
void
IoWorker::readIncomingData(int fd) {
char buffer[Const::MaxBuffer];
memset(buffer, 0, sizeof buffer);
ssize_t totalBytes = 0;
for (;;) {
ssize_t bytes;
bytes = recv(fd, buffer + totalBytes, Const::MaxBuffer - totalBytes, 0);
if (bytes == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
cout << "Received " << buffer << endl;
} else {
throw std::runtime_error(strerror(errno));
}
break;
}
else if (bytes == 0) {
cout << "Peer has been shutdowned" << endl;
break;
}
else {
totalBytes += bytes;
if (totalBytes >= Const::MaxBuffer) {
cerr << "Too long packet" << endl;
break;
}
}
}
}
void
IoWorker::run() {
struct epoll_event events[Const::MaxEvents];
for (;;) {
int ready_fds;
switch(ready_fds = epoll_wait(epoll_fd, events, Const::MaxEvents, 100)) {
case -1:
break;
case 0:
if (!mailbox.isEmpty()) {
int *fd = mailbox.clear();
struct epoll_event event;
event.events = EPOLLIN;
event.data.fd = *fd;
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, *fd, &event);
delete fd;
}
break;
default:
for (int i = 0; i < ready_fds; ++i) {
const struct epoll_event *event = events + i;
readIncomingData(event->data.fd);
}
break;
}
}
}
Listener::Listener(const Address& address)
: addr_(address)
, listen_fd(-1)
{
}
bool Listener::bind() {
struct addrinfo hints;
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
hints.ai_flags = AI_PASSIVE;
hints.ai_protocol = 0;
auto host = addr_.host();
if (host == "*") {
host = "0.0.0.0";
}
/* We rely on the fact that a string literal is an lvalue const char[N] */
static constexpr size_t MaxPortLen = sizeof("65535");
char port[MaxPortLen];
std::fill(port, port + MaxPortLen, 0);
std::snprintf(port, MaxPortLen, "%d", addr_.port());
struct addrinfo *addrs;
TRY(::getaddrinfo(host.c_str(), port, &hints, &addrs));
int fd = -1;
for (struct addrinfo *addr = addrs; addr; addr = addr->ai_next) {
fd = ::socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol);
if (fd < 0) continue;
if (::bind(fd, addr->ai_addr, addr->ai_addrlen) < 0) {
close(fd);
continue;
}
TRY(::listen(fd, Const::MaxBacklog));
}
listen_fd = fd;
for (int i = 0; i < 4; ++i) {
auto wrk = std::unique_ptr<IoWorker>(new IoWorker);
wrk->start();
ioGroup.push_back(std::move(wrk));
}
return true;
}
void Listener::run() {
for (;;) {
struct sockaddr_in peer_addr;
socklen_t peer_addr_len = sizeof(peer_addr);
int client_fd = TRY_RET(::accept(listen_fd, (struct sockaddr *)&peer_addr, &peer_addr_len));
make_non_blocking(client_fd);
char peer_host[NI_MAXHOST];
if (getnameinfo((struct sockaddr *)&peer_addr, peer_addr_len, peer_host, NI_MAXHOST, nullptr, 0, 0) == 0) {
Address addr = Address::fromUnix((struct sockaddr *)&peer_addr);
Peer peer(addr, peer_host);
std::cout << "New peer: " << peer << std::endl;
dispatchConnection(client_fd);
}
}
}
void Listener::dispatchConnection(int fd) {
size_t start = fd % 4;
/* Find the first available worker */
size_t current = start;
for (;;) {
auto& mailbox = ioGroup[current]->mailbox;
if (mailbox.isEmpty()) {
int *message = new int(fd);
int *old = mailbox.post(message);
assert(old == nullptr);
return;
}
current = (current + 1) % 4;
if (current == start) {
break;
}
}
/* We did not find any available worker, what do we do ? */
}
} // namespace Tcp
} // namespace Net
/* listener.h
Mathieu Stefani, 12 August 2015
A TCP Listener
*/
#pragma once
#include "net.h"
#include "mailbox.h"
#include <vector>
#include <memory>
#include <thread>
namespace Net {
namespace Tcp {
class IoWorker {
public:
Mailbox<int> mailbox;
IoWorker();
~IoWorker();
void start();
private:
int epoll_fd;
std::unique_ptr<std::thread> thread;
void readIncomingData(int fd);
void run();
};
class Listener {
public:
Listener(const Address& address);
bool bind();
void run();
private:
Address addr_;
int listen_fd;
std::vector<std::unique_ptr<IoWorker>> ioGroup;
void dispatchConnection(int fd);
};
} // namespace Tcp
} // namespace Net
/* mailbox.h
Mathieu Stefani, 12 August 2015
Copyright (c) 2014 Datacratic. All rights reserved.
A simple lock-free Mailbox implementation
*/
#pragma once
#include <atomic>
#include <stdexcept>
template<typename T>
class Mailbox {
public:
const T *get() const {
if (isEmpty()) {
throw std::runtime_error("Can not retrieve mail from empty mailbox");
}
return data.load();
}
T *post(T *newData) {
T *old = data.load();
while (!data.compare_exchange_weak(old, newData))
{ }
return old;
}
T *clear() {
return data.exchange(nullptr);
}
bool isEmpty() const {
return data == nullptr;
}
private:
std::atomic<T *> data;
};
/* net.cc
Mathieu Stefani, 12 August 2015
*/
#include "net.h"
#include "common.h"
#include <stdexcept>
#include <limits>
#include <cstring>
#include <fcntl.h>
#include <netinet/in.h>
#include <arpa/inet.h>
using namespace std;
namespace Net {
bool make_non_blocking(int sfd)
{
int flags = fcntl (sfd, F_GETFL, 0);
if (flags == -1) return false;
flags |= O_NONBLOCK;
int ret = fcntl (sfd, F_SETFL, flags);
if (ret == -1) return false;
return true;
}
Ipv4::Ipv4(uint8_t a, uint8_t b, uint8_t c, uint8_t d)
: a(a)
, b(b)
, c(c)
, d(d)
{ }
Ipv4 Ipv4::any() {
return Ipv4(0, 0, 0, 0);
}
std::string Ipv4::toString() const {
static constexpr size_t MaxSize = sizeof("255") * 4 + 3 + 1; /* 4 * 255 + 3 * dot + \0 */
char buff[MaxSize];
snprintf(buff, MaxSize, "%d.%d.%d.%d", a, b, c, d);
return std::string(buff);
}
Address::Address()
: host_("")
, port_(0)
{ }
Address::Address(std::string host, Port port)
: host_(std::move(host))
, port_(port)
{ }
Address::Address(std::string addr)
{
auto pos = addr.find(':');
if (pos == std::string::npos)
throw std::invalid_argument("Invalid address");
std::string host = addr.substr(0, pos);
char *end;
const std::string portPart = addr.substr(pos + 1);
long port = strtol(portPart.c_str(), &end, 10);
if (*end != 0 || port > std::numeric_limits<Port>::max())
throw std::invalid_argument("Invalid port");
host_ = std::move(host);
port_ = port;
}
Address::Address(Ipv4 ip, Port port)
: host_(ip.toString())
, port_(port)
{ }
Address
Address::fromUnix(struct sockaddr* addr) {
struct sockaddr_in *in_addr = reinterpret_cast<struct sockaddr_in *>(addr);
std::string host = TRY_RET(inet_ntoa(in_addr->sin_addr));
int port = in_addr->sin_port;
return Address(std::move(host), port);
}
std::string Address::host() const {
return host_;
}
Port Address::port() const {
return port_;
}
} // namespace Net
/* net.h
Mathieu Stefani, 12 August 2015
Network utility classes
*/
#pragma once
#include <string>
#include <sys/socket.h>
namespace Net {
typedef uint16_t Port;
bool make_non_blocking(int fd);
class Ipv4 {
public:
Ipv4(uint8_t a, uint8_t b, uint8_t c, uint8_t d);
static Ipv4 any();
std::string toString() const;
private:
uint8_t a;
uint8_t b;
uint8_t c;
uint8_t d;
};
class Address {
public:
Address();
Address(std::string host, Port port);
Address(std::string addr);
Address(Ipv4 ip, Port port);
Address(const Address& other) = default;
Address(Address&& other) = default;
static Address fromUnix(struct sockaddr *addr);
std::string host() const;
Port port() const;
private:
std::string host_;
Port port_;
};
} // namespace Net
/* peer.cc
Mathieu Stefani, 12 August 2015
*/
#include "peer.h"
#include <iostream>
namespace Net {
using namespace std;
Peer::Peer()
{ }
Peer::Peer(const Address& addr, const string& host)
: addr(addr)
, host(host)
{ }
Address
Peer::address() const {
return addr;
}
string
Peer::hostname() const {
return host;
}
std::ostream& operator<<(std::ostream& os, const Peer& peer) {
const auto& addr = peer.address();
os << "(" << addr.host() << ", " << addr.port() << ") [" << peer.hostname() << "]";
return os;
}
} // namespace Net
/* peer.h
Mathieu Stefani, 12 August 2015
A class representing a TCP Peer
*/
#pragma once
#include "net.h"
#include <string>
#include <iostream>
namespace Net {
class Peer {
public:
Peer();
Peer(const Address& addr, const std::string& hostname);
Address address() const;
std::string hostname() const;
private:
Address addr;
std::string host;
};
std::ostream& operator<<(std::ostream& os, const Peer& peer);
} // namespace Net
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment