SocketReactor performance with version 12.4 is far below same code with version 11.6

[zzeng99]

socket send performance with version 12.4 only achieved 5% speed of same code with version 11.6, anything wrong ?

[micheleselea]

mmm what do you mean 5%? do you have a sample code?

[micheleselea]

I think I can't access that page or I can't download it

[micheleselea]

I think you can grad and drop file directly here

[aleks-f]

@zzeng99 nobody can access your link. If you want this issue to be addressed, provide the code somewhere where it can be accessed.

I'm puzzled about the secrecy - you publicly report a problem about publicly available code, then got through effort to hide the code exhibiting the problem. why?

[aleks-f]

Can't reproduce:

void SocketReactorTest::testDataCollection()
{
	SocketAddress ssa;
	ServerSocket ss(ssa);
	SocketReactor reactor;
	SocketAcceptor<DataServiceHandler> acceptor(ss, reactor);
	std::string data0("{"
					  "  \"src\":\"127.0.0.1\","
					  "  \"id\":\"test0\","
					  "  \"ts\":\"1524864651000001\","
					  "  \"data\":123"
					  "}\n");

	std::string data1("{"
					  "  \"src\":\"127.0.0.1\","
					  "  \"id\":\"test1\","
					  "  \"ts\":\"1524864651123456\","
					  "  \"data\":"
					  "  ["
					  "   {"
					  "     \"tag1\":"
					  "     ["
					  "      {\"val1\":123},"
					  "      {\"val2\":\"abc\"}"
					  "     ]"
					  "   }"
					  "  ]"
					  "}\n");

	std::string data2 = "{"
						"  \"src\":\"127.0.0.1\","
						"  \"id\":\"test2\","
						"  \"ts\":\"1524864652654321\","
						"  \"data\":"
						"  ["
						"    {"
						"     \"tag1\":"
						"     ["
						"      {"
						"       \"val1\":123,"
						"       \"val2\":\"abc\","
						"       \"val3\":42.123"
						"      },"
						"      {"
						"       \"val1\":987,"
						"       \"val2\":\"xyz\","
						"       \"val3\":24.321"
						"      }"
						"     ],"
						"     \"tag2\":"
						"     ["
						"      {"
						"       \"val1\":42.123,"
						"       \"val2\":123,"
						"       \"val3\":\"abc\""
						"      },"
						"      {"
						"       \"val1\":24.321,"
						"       \"val2\":987,"
						"       \"val3\":\"xyz\""
						"      }"
						"    ]"
						"   }"
						" ]"
						"}\n";

	Thread thread;
	thread.start(reactor);

	SocketAddress sa("127.0.0.1", ss.address().port());
	StreamSocket sock(sa);

	Stopwatch sw;
	sw.start();
	sock.sendBytes(data0.data(), static_cast<int>(data0.size()));
	sock.sendBytes(data1.data(), static_cast<int>(data1.size()));
	sock.sendBytes(data2.data(), static_cast<int>(data2.size()));
	while ((DataServiceHandler::_data.size() < 4)) Thread::sleep(1);
	sw.stop();
	std::cout << "Elapsed time: " << sw.elapsed() << std::endl;

	reactor.stop();
	thread.join();

	assertTrue (DataServiceHandler::_data.size() == 4);
	data0.erase(data0.size() - 1);
	assertTrue (DataServiceHandler::_data[0] == data0);
	data1.erase(data1.size() - 1);
	assertTrue (DataServiceHandler::_data[1] == data1);
	data2.erase(data2.size() - 1);
	assertTrue (DataServiceHandler::_data[2] == data2);
	assertTrue (DataServiceHandler::_data[3].empty());
}

• devel: 1175, 1190, 1479, 1491, 1437
• 1.12.4: 1488, 1466, 1384, 1422, 1133
• 1.12.5: 1256, 1415, 1396, 1198, 1231
• 1.11.6: 1240, 1412, 1336, 1339, 1392

[micheleselea]

windows/linux?

[micheleselea]

Can be something about the wakeUp function that is not called?

[aleks-f]

windows/linux?

linux

[aleks-f]

Can be something about the wakeUp function that is not called?

Impossible to tell without seeing his code

[micheleselea]

yes, I was just guessing about my experience with SocketReactor

[zzeng99]

i had uploaded again, try another link please~~

[aleks-f]

@zzeng99 sorry, but I have no time to troubleshoot your application. Perhaps someone else can do it, or if you can provide a poco test case that exhibits this problem, I'll look into it. So far, I was not able to reproduce it.

See this for an example of what I'm looking for.

[micheleselea]

Adding/removing write read error handler are controlled by a lock in pollset.
When you are in .poll(..) function (at least in windows platform) the pollset is locket so you can't add or remove that kind of handler, you have to wait for pollset to exit poll. So to avoid locks you should add your read/write/error handlers in onidle ontimeout or onbusy of reactor/sockethandler because that way you are outside the poll function

[micheleselea]

I understand your question. You have to keep in mind the PollSet is changed between the version you have, so probably you should take some time to understand where poll and addevent is blocking each other and understand why idle timeout and busy can do the job because are outside the lock. So I think yes you can, but you have to have a look and dig into socketreactor code

[zzeng99]

#include <string>
#include <iostream>
#include "Poco/NObserver.h"
#include "Poco/Thread.h"
#include "Poco/FIFOBuffer.h"
#include "Poco/Delegate.h"
#include "Poco/AutoPtr.h"
#include "Poco/Timespan.h"
#include "Poco/DateTimeFormat.h"
#include "Poco/Net/SocketAcceptor.h"
#include "Poco/Net/SocketNotification.h"

using namespace Poco::Net;
using namespace Poco;
using namespace std;

#define BUFFER_SIZE 2048
int loop = 5000;
FIFOBuffer *pOut = nullptr;

class TcpClient : public Runnable {

public:
    TcpClient(string ip, int port) {
        ss.connect(SocketAddress(ip, port));
    }

    ~TcpClient() {
        ss.close();
    }

protected:

    void run() {

        Timespan span(10000);
        for (int i = 0; i < loop; i++) {

            ss.receiveBytes(buffer, sizeof(buffer));
            ss.sendBytes(buffer, sizeof(buffer));
        }

        delete this;
    }

private:
    StreamSocket ss;
    char buffer[BUFFER_SIZE];
};


class ServerServiceHandler {
public:

    ServerServiceHandler(StreamSocket &socket, SocketReactor &reactor) :
            _socket(socket),
            _reactor(reactor),
            counter(0),
            _fifoIn(BUFFER_SIZE, true) {

        cout << "new connection from " << _socket.peerAddress() << endl;
        pOut = new FIFOBuffer(BUFFER_SIZE, true);

        pOut->readable += delegate(this, &ServerServiceHandler::onFIFOOutReadable);
        _fifoIn.writable += delegate(this, &ServerServiceHandler::onFIFOInWritable);

        _reactor.addEventHandler(_socket, NObserver<ServerServiceHandler, ReadableNotification>(*this,
                                                                                                &ServerServiceHandler::onSocketReadable));
    }

    ~ServerServiceHandler() {

        _reactor.removeEventHandler(_socket, NObserver<ServerServiceHandler, ReadableNotification>(*this,
                                                                                                   &ServerServiceHandler::onSocketReadable));
        _reactor.removeEventHandler(_socket, NObserver<ServerServiceHandler, WritableNotification>(*this,
                                                                                                   &ServerServiceHandler::onSocketWritable));

        pOut->readable -= delegate(this, &ServerServiceHandler::onFIFOOutReadable);
        _fifoIn.writable -= delegate(this, &ServerServiceHandler::onFIFOInWritable);

        delete pOut;
    }

    void onFIFOOutReadable(bool &b) {

        if (b) {
            counter++;
            Timestamp ts;
            _reactor.addEventHandler(_socket, NObserver<ServerServiceHandler, WritableNotification>(*this,
                                                                                                    &ServerServiceHandler::onSocketWritable));
            if (ts.elapsed() > 100)
                cout << counter << " : addEventHandler cost " << ts.elapsed() << " us" << endl;
        } else {
            _reactor.removeEventHandler(_socket, NObserver<ServerServiceHandler, WritableNotification>(*this,
                                                                                                       &ServerServiceHandler::onSocketWritable));
        }
    }


    void onFIFOInWritable(bool &b) {

        if (b) { // not to writeable, means have msg for socket to receive
            _reactor.addEventHandler(_socket, NObserver<ServerServiceHandler, ReadableNotification>(*this,
                                                                                                    &ServerServiceHandler::onSocketReadable));
        } else { // writeable to not, means buffer is full & can't read msg
            _fifoIn.resize(_fifoIn.size() + BUFFER_SIZE, true);
            _reactor.addEventHandler(_socket, NObserver<ServerServiceHandler, ReadableNotification>(*this,
                                                                                                    &ServerServiceHandler::onSocketReadable));
        }
    }


    void onSocketReadable(const AutoPtr<ReadableNotification> &pNf) {

        int len = _socket.receiveBytes(_fifoIn);

        if (len > 0) {

            UInt32 bufferLen = _fifoIn.used();
            _fifoIn.drain(bufferLen);

        } else
            delete this;
    }


    void onSocketWritable(const AutoPtr<WritableNotification> &pNf) {

        _socket.sendBytes(*pOut);
    }

private:
    StreamSocket _socket;
    SocketReactor &_reactor;
    FIFOBuffer _fifoIn;
    int counter;
};

int main() {

    Timespan timeout(1000);
    // set-up a server socket
    ServerSocket svs(9977);
    // set-up a SocketReactor using timeout 1 millisecond...
    SocketReactor reactor(timeout);
    // ... and a SocketAcceptor
    SocketAcceptor<ServerServiceHandler> acceptor2(svs, reactor);
    // run the reactor in its own thread so that we can wait for
    // a termination request
    Thread threadServer;
    threadServer.start(reactor);

    TcpClient *client = new TcpClient("localhost", 9977);
    Thread threadClient;
    threadClient.start(*client);

    Thread::sleep(1000);

    char msg[1024] = {0};

    Timestamp ts;
    cout << "begin test ....." << endl;
    for (int i = 0; i < loop; i++) {

        usleep(1);

        if (pOut != nullptr) {
            pOut->write(msg, 1024);
        }

    }
    cout << "end test , echo " << loop << " msg, total cost " << ts.elapsed() / (1000 * 1000) << " seconds" << endl;

    reactor.stop();

    return 0;
    
}

[micheleselea]

I think the problem is the lack of idle, timeout and busy callback and the lack of call to wakUp thread.
I guess if you use
Timespan timeout(2000);
the code run slower.
You have to keep in mind that when reactor is empty it runs idle an than a thread->trysleep
I think you should do a reactor.wakeUp() at the end of ServerServiceHandler constructor
but that works for 1 connection only. When you have more connections you have to deal with timeout and busy callback too, because you have to add readable error and writable when you are not in poll otherwise is blocking.
Have a look to the code and you have to understand how exactly PollSet works and reactor works

[zzeng99]

	PollSet::SocketModeMap poll(const Poco::Timespan& timeout)
	{
		PollSet::SocketModeMap result;
		Poco::Timespan remainingTime(timeout);
		int rc;

		do
		{
			
			Poco::Timestamp start;
			int sz = _events.size();
			
			rc = epoll_wait(_epollfd, &_events[0],
				sz, static_cast<int>(remainingTime.totalMilliseconds()));
			if (rc == 0) return result;
			
			// if we are hitting the events limit, resize it; even without resizing, the subseqent
			// calls would round-robin through the remaining ready sockets, but it's better to give
			// the call enough room once we start hitting the boundary
			if (rc >= sz)
			{
				ScopedLock lock(_mutex);
				_events.resize(sz*2);
			}
			else if (rc < 0)
			{
				// if interrupted and there's still time left, keep waiting
				if (SocketImpl::lastError() == POCO_EINTR)
				{
					Poco::Timestamp end;
					Poco::Timespan waited = end - start;
					if (waited < remainingTime)
					{
						remainingTime -= waited;
						continue;
					}
				}
				else SocketImpl::error();
			}
		}
		while (false);
		
		ScopedLock lock(_mutex);

		for (int i = 0; i < rc; i++)
		{
			if (_events[i].data.ptr) // skip eventfd
			{
				SocketMap::iterator it = _socketMap.find(_events[i].data.ptr);
				if (it != _socketMap.end())
				{
					if (_events[i].events & EPOLLIN)
						result[it->second.first] |= PollSet::POLL_READ;
					if (_events[i].events & EPOLLOUT)
						result[it->second.first] |= PollSet::POLL_WRITE;
					if (_events[i].events & EPOLLERR)
						result[it->second.first] |= PollSet::POLL_ERROR;
				}
			}
			else if (_events[i].events & EPOLLIN) // eventfd signaled
			{
				uint64_t val;
#ifdef WEPOLL_H_
				if (_pSocket && _pSocket->available())
					_pSocket->impl()->receiveBytes(&val, sizeof(val));
#else
				read(_eventfd, &val, sizeof(val));
#endif
			}
		}
		return result;
	}

[zzeng99]

@aleks-f it seems PollSet::poll locked some unnessasery code, i changed the code as above , and it works. now version 12.4 performance same as version 11.6

[micheleselea]

mmm, I think if you lock this

if (rc >= _events.size()) {
				ScopedLock lock(_mutex);
				_events.resize(_events.size()*2);
			}

you probably should lock even the code above

rc = epoll_wait(_epollfd, &_events[0],
				static_cast<int>(_events.size()), static_cast<int>(remainingTime.totalMilliseconds()));

The only way I found to avoid possible time waste is to make the addEventHandler outside the poll code, so in onBusy, onTimeout or in onIdle functions.
And probably you need to call wakeUp to wake up thread sleeping or waiting in epoll_wait.
On windows system the wepoll can be wakeup only with a call to
PostQueuedCompletionStatus(_epollfd, 0, 0, NULL);
if you don't want to have a tcp port open in listen just to do
StreamSocket ss(SocketAddress("127.0.0.1", _port));
as in the original code

[zzeng99]

epoll_wait is thread safe.
call wakeup at constructor does not help
why move addEventHandler in onTimeout ? that is too strange...
just paste your complete code, if it works...

[micheleselea]

Because if you have an empty reactor, suppose you are in the trysleep function (suppose just as example 60 seconds timeout), if a new client arrive and you create the handler, you will never be able to addEventHandler until epoll_wait terminates. That in the current implementation with lock on all the poll function.
Anyway I know that epoll_wait is thread safe but you are filling epoll_wait with _events that can be changed from outside by addEventHandler calls, that why I think the poll function is all locked in the original implementation