Multithreaded SOCKS 4 server on Qt

From time to time on the forums of the RuNet in the Qt branch there are questions related to the programming of network applications. One of the problems that torment these people is the approach to server organization. Usually list three approaches:

• single-line asynchronous;
• multi-threaded; create stream per connection;
• multi-threaded, with a thread pool on QThreadPool and QRunnable .

When you talk about a fixed number of worker threads with your event-handling cycle, you are asked to give an example. Below is an example of a server with a pool of threads, each of which has its own event-handling cycle.

Characters:

• class Server , joins connections and distributes tasks to workers;
• the Worker class, instances of which will create instances of the Client class in the workflows;
• Client class encapsulating client requests and implementing SOCKS 4

The simplest of this trinity is the Worker, it is inherited from QObject and implements only one client creation function and is purely for the “correct” use of threads :

class Worker: public QObject { Q_OBJECT public: Q_INVOKABLE void addClient(qintptr socketDescriptor); }; 

 void Worker::addClient(qintptr socketDescriptor) { new Client(socketDescriptor, this); } 

The server is just as simple:

 class Server: public QTcpServer { Q_OBJECT public: Server(size_t threads = 4, QObject * parent = nullptr); ~Server(); protected: virtual void incomingConnection(qintptr socketDescriptor); private: void initThreads(); private: size_t m_threadCount; QVector<QThread*> m_threads; QVector<Worker*> m_workers; size_t m_rrcounter; }; 

 Server::Server(size_t threads, QObject * parent) : QTcpServer(parent), m_threadCount(threads), m_rrcounter(0) { initThreads(); } Server::~Server() { for(QThread* thread: m_threads) { thread->quit(); thread->wait(); } } void Server::initThreads() { for (size_t i = 0; i < m_threadCount; ++i) { QThread* thread = new QThread(this); Worker* worker = new Worker(); worker->moveToThread(thread); connect(thread, &QThread::finished, worker, &QObject::deleteLater); m_threads.push_back(thread); m_workers.push_back(worker); thread->start(); } } void Server::incomingConnection(qintptr socketDescriptor) { Worker* worker = m_workers[m_rrcounter % m_threadCount]; ++m_rrcounter; QMetaObject::invokeMethod(worker, "addClient", Qt::QueuedConnection, Q_ARG(qintptr, socketDescriptor)); } 

In the constructor, it creates threads, workers, and moves workers into threads. Each new connection is passed to the worker. He chooses a worker to “honor”, ​​that is, according to Round-robin .

SOCKS 4 is a very simple protocol, you only need:

1. read IP address, port number;
2. establish a connection with the "world";
3. send a message to the client that the request is confirmed;
4. send data from one socket to another until someone closes the connection.

 class Client: public QObject { Q_OBJECT public: Client(qintptr socketDescriptor, QObject* parent = 0); public slots: void onRequest(); void client2world(); void world2client(); void sendSocksAnsver(); void onClientDisconnected(); void onWorldDisconnected(); private: void done(); private: QTcpSocket m_client; QTcpSocket m_world; }; 

 namespace { #pragma pack(push, 1) struct socks4request { uint8_t version; uint8_t command; uint16_t port; uint32_t address; uint8_t end; }; struct socks4ansver { uint8_t empty = 0; uint8_t status; uint16_t field1 = 0; uint32_t field2 = 0; }; #pragma pack(pop) enum SocksStatus { Granted = 0x5a, Failed = 0x5b, Failed_no_identd = 0x5c, Failed_bad_user_id = 0x5d }; } Client::Client(qintptr socketDescriptor, QObject* parent) : QObject(parent) { m_client.setSocketDescriptor(socketDescriptor); connect(&m_client, &QTcpSocket::readyRead, this, &Client::onRequest); connect(&m_client,&QTcpSocket::disconnected, this, &Client::onClientDisconnected); connect(&m_world, &QTcpSocket::connected, this, &Client::sendSocksAnsver); connect(&m_world, &QTcpSocket::readyRead, this, &Client::world2client); connect(&m_world,&QTcpSocket::disconnected, this, &Client::onWorldDisconnected); } void Client::onRequest() { QByteArray request = m_client.readAll(); socks4request* header = reinterpret_cast<socks4request*>(request.data()); #if Q_BYTE_ORDER == Q_LITTLE_ENDIAN const QHostAddress address(qFromBigEndian(header->address)); #else const QHostAddress address(header->address); #endif #if Q_BYTE_ORDER == Q_LITTLE_ENDIAN const uint16_t port = qFromBigEndian(header->port); #else const uint16_t port = header->port; #endif //qDebug()<<"connection:"<<address<<"port:"<<port; m_world.connectToHost(address, port); disconnect(&m_client, &QTcpSocket::readyRead, this, &Client::onRequest); connect(&m_client, &QTcpSocket::readyRead, this, &Client::client2world); } void Client::sendSocksAnsver() { socks4ansver ans; ans.status = Granted; m_client.write(reinterpret_cast<char*>(&ans), sizeof(ans)); m_client.flush(); } void Client::client2world() { m_world.write(m_client.readAll()); } void Client::world2client() { m_client.write(m_world.readAll()); } void Client::onClientDisconnected() { m_world.flush(); done(); } void Client::onWorldDisconnected() { m_client.flush(); done(); } void Client::done() { m_client.close(); m_world.close(); deleteLater(); } 

Epoll and Qt

A cry is like thunder:
- Give people rum
Need any
People drink rum!

If we compile the previous code and run it through strace -f , we will see calls to poll. There will definitely be someone who will say his weighty "fi", saying that with an epoll there will be "well, finally, a rocket."
')
Qt has a QAbstractEventDispatcher class that allows you to define your own event dispatcher. Naturally, there were good people who made and posted dispatchers with different backends. Here is a small list of them:

• connectedtable / qeventdispatcher_epoll ;
• sjinks / qt_eventdispatcher_epoll ;
• sjinks / qt_eventdispatcher_libev ;
• sjinks / qt_eventdispatcher_libevent
• svalaskevicius / qt-event-dispatcher-libuv

When using our dispatcher, we register in main.cpp

 QCoreApplication::setEventDispatcher(new QEventDispatcherEpoll); QCoreApplication app(argc, argv) 

and the initThreads method on the server becomes:

 void Server::initThreads() { for (size_t i = 0; i < m_threadCount; ++i) { QThread* thread = new QThread(this); thread->setEventDispatcher(new QEventDispatcherEpoll); Worker* worker = new Worker(); worker->moveToThread(thread); connect(thread, &QThread::finished, worker, &QObject::deleteLater); m_threads.push_back(thread); m_workers.push_back(worker); thread->start(); } } 

And if we run strace again, we will see the cherished function calls with the epoll_ prefix.

findings

The conclusions are purely pragmatic.

If you are an application programmer and you have no tasks from the category of “big” data or highload in Bunin, then write on what you want and how you can. The task of the application programmer to produce a product of a certain quality, spending a certain amount of resources. Otherwise, one only sockets with epoll will not do.

PS

Source codes are available on GitHub .