Reading from sockets in a multithreaded program

[danadam]

I'm not 100% sure of this, cause I have no way to check it now, but:

You create QTcpSocket in Connection constructor, so it belongs to main thread. In Connection thread you call waitForRead() methods, so Connection thread is blocked until some data is available in socket. The following part is the one I'm not sure. I think that even when you use waitFor...() methods in Connection thread, the QSocket is still processed by the thread it lives in. So in order to receive any data (and as a result to unblock Connection thread) the main thread has to do some job. But it may happen that it will be blocked on acquire() and you get deadlock. Main thread waits for Connection thread to release semaphore and Connection thread waits for Main thread to process socket and deliver some data.

Try to move QSocket creation to run() method of Connection class and see if it helps.

[KoosKoets]

Main thread waits for Connection thread to release semaphore and Connection thread waits for Main thread to process socket and deliver some data.

Try to move QSocket creation to run() method of Connection class and see if it helps.

This seems a plausible explanation. However, after trying this I still mange get the 'deadlock'. Only in about 5% of my tries, all data is received without problems.

[danadam]

I think I found the reason. 58th line of Connection code in the first post is:

lBufLength += mSocket->bytesAvailable();

lBufLength += mSocket->bytesAvailable();

To copy to clipboard, switch view to plain text mode 

I guess it should be:

lBufLength = mSocket->bytesAvailable();

lBufLength = mSocket->bytesAvailable();

To copy to clipboard, switch view to plain text mode 

[KoosKoets]

I guess it should be:

lBufLength = mSocket->bytesAvailable();

lBufLength = mSocket->bytesAvailable();

To copy to clipboard, switch view to plain text mode 

Good catch: it is indeed wrong. However, my deadlock remains...

[high_flyer]

Good catch: it is indeed wrong. However, my deadlock remains...

Why don't you try tryLock()?
It is easy, and it will tell you on the spot if its a mutex deadlock that you have...

[KoosKoets]

Why don't you try tryLock()?

I did try this, but still my problem remains.

I've stripped my code even further. Upon reception of a packet it is deleted right away (it is not placed in a mutex guarded QQueue, no semaphore release, no semaphore acquire): just reception of data and deletion. The creation of the QTcpSocket is done in the run() function. My problem remains...

It is like the waitForReadyRead(-1) sometimes doesn't wake up even when data is present at the socket.

[moowy]

I have a problem with using Qt networking. I want to establish a connection between a server and clients. The client is using a chalkboard and is using server to send coordinates of the lines to draw to them. For the drawing part I am using Irrlecht (because the program is not going to be the chalboard (this is just for testing), but a real 3d environment.

I have problem getting the communication to work. I would really appreciate if someone could look at the code (I attached) and please give me some clue.

On the server side I have:

Server::Server(QObject *parent) : QTcpServer(parent)
{
	/* maximum number of players */
	setMaxPendingConnections(8);
	m_connectedClients=-1; // none connected clients
	connect(this,SIGNAL(newConnection()),SLOT(newConnections()));
 
}
void Server::send(int socket,int x1, int y1, int x2, int y2)
{
	// I use qhash to keep information about all clients and socketDescriptors
	QHashIterator<int, Server_thread*> i(m_clients);
	while (i.hasNext()) {
		i.next();
		if(i.key() != socket) // don't send this to the client which send the info
			i.value()->send(x1,y1,x2,y2);
	} 
}
 
/* accept incoming connections */
void Server::newConnections()
{
	while(hasPendingConnections()) {
		QTcpSocket *socket = nextPendingConnection();
		Server_thread *thread = new Server_thread(socket, this);
		// add the client and socketDescriptor
                m_clients[socket->socketDescriptor()] = thread;
		/* delete the thread that has finished */
		connect(thread, SIGNAL(finished()), thread, SLOT(deleteLater()));
		thread->start();
	}
}

Server::Server(QObject *parent) : QTcpServer(parent)
{
	/* maximum number of players */
	setMaxPendingConnections(8);
	m_connectedClients=-1; // none connected clients
	connect(this,SIGNAL(newConnection()),SLOT(newConnections()));

}
void Server::send(int socket,int x1, int y1, int x2, int y2)
{
	// I use qhash to keep information about all clients and socketDescriptors
	QHashIterator<int, Server_thread*> i(m_clients);
	while (i.hasNext()) {
		i.next();
		if(i.key() != socket) // don't send this to the client which send the info
			i.value()->send(x1,y1,x2,y2);
	} 
}

/* accept incoming connections */
void Server::newConnections()
{
	while(hasPendingConnections()) {
		QTcpSocket *socket = nextPendingConnection();
		Server_thread *thread = new Server_thread(socket, this);
		// add the client and socketDescriptor
                m_clients[socket->socketDescriptor()] = thread;
		/* delete the thread that has finished */
		connect(thread, SIGNAL(finished()), thread, SLOT(deleteLater()));
		thread->start();
	}
}

To copy to clipboard, switch view to plain text mode 

The thread for clients...

Server_thread::Server_thread(QTcpSocket *socket, Server *parent)
    : QThread(parent)
{
	m_parent = parent;
	m_socket = socket;
	connect(this,SIGNAL(address(QString)),this,SLOT(new_connection(QString)));
	connect(m_socket,SIGNAL(readyRead()),this,SLOT(read()));
	m_socketDescriptor = m_socket->socketDescriptor();
 
	emit address(m_socket->peerAddress().toString());
 
	/* number of clients++ */
	m_parent->incNrClients();
 
	nextBlockSize=0;
}
 
 
void Server_thread::run()
{
	// just to keep the thread active...
	while(m_socket && m_socket->state() == QAbstractSocket::ConnectedState) {
 
	}
 
	m_socket->disconnectFromHost();
	m_socket->waitForDisconnected(1000);
}
 
void Server_thread::read()
{
	QDataStream in(m_socket);
	in.setVersion(QDataStream::Qt_4_2);
	/* read data */
	if(nextBlockSize==0) {
		if(m_socket->bytesAvailable() < sizeof(quint16))
			return;
 
		in >> nextBlockSize;
	}
	if(m_socket->bytesAvailable() < nextBlockSize)
		return;
 
	QString koordinate;
	int x1,y1,x2,y2;
 
	in >> koordinate;
 
	sscanf(koordinate.toStdString().c_str(),"%d %d %d %d",&x1,&y1,&x2,&y2);
 
	m_parent->send(m_socketDescriptor,x1,y1,x2,y2);	
}
 
void Server_thread::send(int x1, int y1, int x2, int y2)
{
	QString string;
	string += QString(to_string<int>(x1).c_str());
	string += " ";
	string += QString(to_string<int>(y1).c_str());
	string += " ";
	string += QString(to_string<int>(x2).c_str());
	string += " ";
	string += QString(to_string<int>(y2).c_str());
 
 
	QByteArray block;
	QDataStream out(&block, QIODevice::WriteOnly);
	out.setVersion(QDataStream::Qt_4_2);
	out << (quint16)0;
	out << string;
	out.device()->seek(0);
	out << (quint16)(block.size() - sizeof(quint16));
 
	m_socket->write(block);
 
}

Server_thread::Server_thread(QTcpSocket *socket, Server *parent)
    : QThread(parent)
{
	m_parent = parent;
	m_socket = socket;
	connect(this,SIGNAL(address(QString)),this,SLOT(new_connection(QString)));
	connect(m_socket,SIGNAL(readyRead()),this,SLOT(read()));
	m_socketDescriptor = m_socket->socketDescriptor();
	
	emit address(m_socket->peerAddress().toString());
	
	/* number of clients++ */
	m_parent->incNrClients();
	
	nextBlockSize=0;
}


void Server_thread::run()
{
	// just to keep the thread active...
	while(m_socket && m_socket->state() == QAbstractSocket::ConnectedState) {
		
	}
	
	m_socket->disconnectFromHost();
	m_socket->waitForDisconnected(1000);
}

void Server_thread::read()
{
	QDataStream in(m_socket);
	in.setVersion(QDataStream::Qt_4_2);
	/* read data */
	if(nextBlockSize==0) {
		if(m_socket->bytesAvailable() < sizeof(quint16))
			return;

		in >> nextBlockSize;
	}
	if(m_socket->bytesAvailable() < nextBlockSize)
		return;

	QString koordinate;
	int x1,y1,x2,y2;
	
	in >> koordinate;

	sscanf(koordinate.toStdString().c_str(),"%d %d %d %d",&x1,&y1,&x2,&y2);

	m_parent->send(m_socketDescriptor,x1,y1,x2,y2);	
}

void Server_thread::send(int x1, int y1, int x2, int y2)
{
	QString string;
	string += QString(to_string<int>(x1).c_str());
	string += " ";
	string += QString(to_string<int>(y1).c_str());
	string += " ";
	string += QString(to_string<int>(x2).c_str());
	string += " ";
	string += QString(to_string<int>(y2).c_str());


	QByteArray block;
	QDataStream out(&block, QIODevice::WriteOnly);
	out.setVersion(QDataStream::Qt_4_2);
	out << (quint16)0;
	out << string;
	out.device()->seek(0);
	out << (quint16)(block.size() - sizeof(quint16));

	m_socket->write(block);

}

To copy to clipboard, switch view to plain text mode 

On the client side I have:

using namespace irr;
using namespace core;
using namespace scene;
using namespace video;
 
//event listener -> this works for sure
class ChalkboardEventReceiver : public IEventReceiver
{
public:
    ChalkboardEventReceiver(Client *c) 
    : mouseDown(false), connection(NULL)
    {
        connection = c;
    }
 
    bool OnEvent(SEvent event)
    {
        if(event.EventType == EET_MOUSE_INPUT_EVENT) {
            if(event.MouseInput.Event == EMIE_LMOUSE_PRESSED_DOWN) {
                x = event.MouseInput.X;
                y = event.MouseInput.Y;
                mouseDown = true;
 
                return true;
            }
            else if(event.MouseInput.Event == EMIE_LMOUSE_LEFT_UP) {
				connection->AddLineLocal(x, y, event.MouseInput.X, event.MouseInput.Y);
                connection->SendLineToServer(x, y, event.MouseInput.X, event.MouseInput.Y);
                mouseDown = false;
 
                return true;
            }
            else if(mouseDown && event.MouseInput.Event == EMIE_MOUSE_MOVED) {
                connection->AddLineLocal(x, y, event.MouseInput.X, event.MouseInput.Y);
                connection->SendLineToServer(x, y, event.MouseInput.X, event.MouseInput.Y);
                x = event.MouseInput.X;
                y = event.MouseInput.Y;
 
                return true;
            }
        }
 
        return false;
    }
 
 
protected:
    s32 x, y;
    bool mouseDown;
    Client *connection;
};
 
int main(int argc, char *argv[])
{
	QApplication app(argc, argv);
 
        Client client;
	ChalkboardEventReceiver receiver(&client);
	client.init(&receiver);
 
	app.connect(&app, SIGNAL(lastWindowClosed()), &app, SLOT(quit()));
	return app.exec();
 
}
 
Client::Client(QWidget *parent) : QWidget(parent)
{
	connect(&m_socket, SIGNAL(readyRead()), this, SLOT(HandlePacket()));
	connect(&m_socket, SIGNAL(error(QAbstractSocket::SocketError)),
             this, SLOT(displayError(QAbstractSocket::SocketError)));
        m_socket.connectToHost("127.0.0.1",3210);
	nextBlockSize = 0;
	// Wait for the init() call
    device = 0;
    // Default 
   driverType = irr::video::EDT_OPENGL;
}
 
void Client::SendLineToServer(s32 x1, s32 y1, s32 x2, s32 y2)
{
 
	QString string;
	string += QString(to_string<int>(x1).c_str());
	string += " ";
	string += QString(to_string<int>(y1).c_str());
	string += " ";
	string += QString(to_string<int>(x2).c_str());
	string += " ";
	string += QString(to_string<int>(y2).c_str());
 
	QByteArray block;
	QDataStream out(&block, QIODevice::WriteOnly);
	out.setVersion(QDataStream::Qt_4_2);
	out << (quint16)0;
	out << string;
	out.device()->seek(0);
 
	//calculate the size of the block
	out << (quint16)(block.size() - sizeof(quint16));
 
	m_socket.write(block);
}
 
void Client::HandlePacket()
{
	QDataStream in(&m_socket);
    in.setVersion(QDataStream::Qt_4_2);
 
	if (nextBlockSize == 0) {
		if (m_socket.bytesAvailable() < (int)sizeof(quint16))
			return;
 
		in >> nextBlockSize;
	}
 
	if (m_socket.bytesAvailable() < nextBlockSize)
		return;
 
	QString string;
	in >> string;
 
	s32 x1,y1,x2,y2;
	const char *str = string.toStdString().c_str();
	sscanf(str,"%d %d %d %d",&x1,&y1,&x2,&y2);
 
	AddLineLocal(x1, y1, x2, y2);
}

using namespace irr;
using namespace core;
using namespace scene;
using namespace video;

//event listener -> this works for sure
class ChalkboardEventReceiver : public IEventReceiver
{
public:
    ChalkboardEventReceiver(Client *c) 
    : mouseDown(false), connection(NULL)
    {
        connection = c;
    }
    
    bool OnEvent(SEvent event)
    {
        if(event.EventType == EET_MOUSE_INPUT_EVENT) {
            if(event.MouseInput.Event == EMIE_LMOUSE_PRESSED_DOWN) {
                x = event.MouseInput.X;
                y = event.MouseInput.Y;
                mouseDown = true;
                
                return true;
            }
            else if(event.MouseInput.Event == EMIE_LMOUSE_LEFT_UP) {
				connection->AddLineLocal(x, y, event.MouseInput.X, event.MouseInput.Y);
                connection->SendLineToServer(x, y, event.MouseInput.X, event.MouseInput.Y);
                mouseDown = false;
                
                return true;
            }
            else if(mouseDown && event.MouseInput.Event == EMIE_MOUSE_MOVED) {
                connection->AddLineLocal(x, y, event.MouseInput.X, event.MouseInput.Y);
                connection->SendLineToServer(x, y, event.MouseInput.X, event.MouseInput.Y);
                x = event.MouseInput.X;
                y = event.MouseInput.Y;
                
                return true;
            }
        }
        
        return false;
    }
    
    
protected:
    s32 x, y;
    bool mouseDown;
    Client *connection;
};

int main(int argc, char *argv[])
{
	QApplication app(argc, argv);
	
        Client client;
	ChalkboardEventReceiver receiver(&client);
	client.init(&receiver);

	app.connect(&app, SIGNAL(lastWindowClosed()), &app, SLOT(quit()));
	return app.exec();

}

Client::Client(QWidget *parent) : QWidget(parent)
{
	connect(&m_socket, SIGNAL(readyRead()), this, SLOT(HandlePacket()));
	connect(&m_socket, SIGNAL(error(QAbstractSocket::SocketError)),
             this, SLOT(displayError(QAbstractSocket::SocketError)));
        m_socket.connectToHost("127.0.0.1",3210);
	nextBlockSize = 0;
	// Wait for the init() call
    device = 0;
    // Default 
   driverType = irr::video::EDT_OPENGL;
}

void Client::SendLineToServer(s32 x1, s32 y1, s32 x2, s32 y2)
{

	QString string;
	string += QString(to_string<int>(x1).c_str());
	string += " ";
	string += QString(to_string<int>(y1).c_str());
	string += " ";
	string += QString(to_string<int>(x2).c_str());
	string += " ";
	string += QString(to_string<int>(y2).c_str());

	QByteArray block;
	QDataStream out(&block, QIODevice::WriteOnly);
	out.setVersion(QDataStream::Qt_4_2);
	out << (quint16)0;
	out << string;
	out.device()->seek(0);

	//calculate the size of the block
	out << (quint16)(block.size() - sizeof(quint16));

	m_socket.write(block);
}
    
void Client::HandlePacket()
{
	QDataStream in(&m_socket);
    in.setVersion(QDataStream::Qt_4_2);
	
	if (nextBlockSize == 0) {
		if (m_socket.bytesAvailable() < (int)sizeof(quint16))
			return;

		in >> nextBlockSize;
	}

	if (m_socket.bytesAvailable() < nextBlockSize)
		return;

	QString string;
	in >> string;

	s32 x1,y1,x2,y2;
	const char *str = string.toStdString().c_str();
	sscanf(str,"%d %d %d %d",&x1,&y1,&x2,&y2);

	AddLineLocal(x1, y1, x2, y2);
}

To copy to clipboard, switch view to plain text mode 

I'm sorry for posting so much code here (I attached the whole source code). I would really appreciate some help.