Two processes communicating...
-
I have two processes which I want to communicate. I had an issue with the processes connecting, where the connection was established but I was not getting any signals firing. This turned out to be because my thread that manages the connection to another thread was now using the Qt::DirectionConnection as the last parameter of the QObject::connect, instead it was using the default which failed.
This problem was the result of my class being re-written and using the moveToThread function.
The problem I've got now is that although one of the processes is writing data to a socket and it appears to be successful as the bytes written is > 0 and matches the packet size, the process listens, I can see the newConnection signal is received and my onNewConnection slot is received:
void clsListener::onNewConnection() { mpClient = mpServer->nextPendingConnection(); if ( mpClient == nullptr || mpClient->isOpen() != true ) { return; } //Connect signals and slots QObject::connect(mpClient, &QAbstractSocket::disconnected ,mpClient, &QObject::deleteLater); QObject::connect(mpClient, &QTcpSocket::readyRead ,this, &clsListener::onDataIn); QObject::connect(mpClient ,QOverload<QAbstractSocket::SocketError>::of(&QAbstractSocket::error) ,this, &clsListener::onErrorOccurred); //Signal new connection, this signal is just incase anything else wants to use it... emit newConnection(*mpClient); }
However I don't get the readyRead signal or at least the onDataIn slot is not being triggered. Can anyone help me find out why?
Here is the listening class clsListener prototype:
class clsListener; typedef std::map<quint16, clsListener*> mpListeners; class clsListener : public QObject { Q_OBJECT private: static mpListeners msmpListeners; static quint16 msuint16NextPort; QTcpSocket* mpClient; QTcpServer* mpServer; quint16 muint16Port; public: clsListener(quint16 uint16Port, QObject* pParent = nullptr); ~clsListener(); static void commonDecode(QAbstractSocket* psckIn); static clsListener* pGetListener(quint16 uint16Port); QTcpSocket* psckClient() { return mpClient; } QTcpServer* psckServer() { return mpServer; } static quint16 uint16NextPort() { return ++clsListener::msuint16NextPort; } quint16 uint16Port() { return muint16Port; } signals: void newConnection(QTcpSocket& robjClient); public slots: void onAcceptError(QAbstractSocket::SocketError socketError); void onDataIn(); void onErrorOccurred(QAbstractSocket::SocketError socketError); void onNewConnection(); };
class clsListener implementation:
/** * File: clsListener.cpp * Notes: Contains implementation of the listener class * History: 2020/11/30 Created by Simon Platten */ #include <utility> #include "clsDebugService.h" #include "clsJSON.h" #include "clsListener.h" //Static initialisation mpListeners clsListener::msmpListeners; quint16 clsListener::msuint16NextPort = 8122; //Will is increment before use /** * @brief clsListener class constructor * @param uint16Port : Port to listen on * @param pParent : Pointer to parent */ clsListener::clsListener(quint16 uint16Port, QObject* pParent) : mpClient(nullptr) , mpServer(nullptr) , muint16Port(uint16Port) { if ( uint16Port == 0 ) { throw "Port invalid!"; } if ( clsListener::pGetListener(uint16Port) != nullptr ) { throw "Port already in use by another listener!"; } clsListener::msmpListeners.insert(std::make_pair(muint16Port, this)); //Connect server signals mpServer = new QTcpServer(pParent); QObject::connect(mpServer, &QTcpServer::acceptError ,this, &clsListener::onAcceptError); QObject::connect(mpServer, &QTcpServer::newConnection ,this, &clsListener::onNewConnection); //Start listening to port if ( !mpServer->listen(QHostAddress::Any, muint16Port) ) { qdbg() << "Unable to lisen to port " << muint16Port; } qdbg() << "This module listening on port: " << muint16Port; } /** * @brief Removes listener */ clsListener::~clsListener() { if ( clsListener::msmpListeners.find(muint16Port) != clsListener::msmpListeners.end() ) { clsListener::msmpListeners.erase(muint16Port); } if ( mpClient != nullptr ) { mpClient->close(); mpClient = nullptr; } if ( mpServer != nullptr ) { QObject::disconnect(mpServer, &QTcpServer::acceptError ,this, &clsListener::onAcceptError); QObject::disconnect(mpServer, &QTcpServer::newConnection ,this, &clsListener::onNewConnection); mpServer->close(); delete mpServer; } } /** * @brief clsListener::commonDecode * @param psckIn : Pointer to socket receiving data */ void clsListener::commonDecode(QAbstractSocket* psckIn) { if ( psckIn->bytesAvailable() == 0 ) { return; } QByteArray arybytMsg(psckIn->readAll()); if ( arybytMsg.isEmpty() == true ) { return; } //Does request start with a 'GET ' prefix? bool blnHTTP = arybytMsg.startsWith(QString("GET ").toLatin1()); int intPos = 0; while( true ) { //Find the start of the JSON object int intOCB = arybytMsg.indexOf(clsJSON::msccOpenCurlyBracket, intPos); if ( intOCB == -1 ) { //No opening curly bracket found, abort return; } //Update reference position intPos = intOCB + 1; //Find the end of the JSON object int intCCB = intOCB, intOCBcnt = 1, intCCBcnt = 0; bool blnFound = false; while( intCCB < arybytMsg.length() ) { //Increment search position intCCB++; if ( arybytMsg[intCCB] == clsJSON::msccOpenCurlyBracket ) { //Another opening curly found, increment count intOCBcnt++; } else if ( arybytMsg[intCCB] == clsJSON::msccCloseCurlyBracket ) { //Increment closing curly count and compare with open count if ( ++intCCBcnt == intOCBcnt ) { //Match found blnFound = true; //Update reference position intPos = intCCB + 1; //Stop this search and continue processing the rest of the packet break; } } } if ( blnFound != true ) { continue; } QString strRx(arybytMsg.mid(intOCB)); //Trancate the request to just the JSON strRx = strRx.mid(0, (intCCB - intOCB) + 1); //Translate request, converting escaped characters strRx = QUrl::fromPercentEncoding(strRx.toLatin1()); #if defined(DEBUG_SOCKETS) qdbg() << QString("clsListener::commonDecode [%1]: %2").arg(strRx.length()).arg(strRx); #endif //Create JSON object clsJSON objMyJSON(strRx); if ( objMyJSON.blnIsValid() != true ) { continue; } //JSON is valid QJsonObject objJSON(objMyJSON.toQJsonObject()); clsJSON::blnDecodeAccordingToType(objJSON); if ( blnHTTP == true ) { //Must be HTTP request, we need the module for the response QJsonObject::iterator itFind = objJSON.find(clsJSON::mscszModule); if ( itFind != objJSON.end() ) { QString strCRLF(clsJSON::mscszCRLF), strModuleName = itFind.value().toString(); QTextStream tsResponse(psckIn); tsResponse.setAutoDetectUnicode(true); //Create thread to deal with request and send response tsResponse << "HTTP/1.0 200 Ok" + strCRLF + "Content-Type: text/html; charset=\"utf-8\"" + strCRLF + strCRLF + + "<h1>Request from " + strModuleName + "</h1>\n"; psckIn->close(); if ( psckIn->state() == QAbstractSocket::UnconnectedState ) { delete psckIn; } } } } } /** * @brief This signal is emitted when accepting a new connection results in * an error. The socketError parameter describes the type of error that * occurred. * @param socketError */ void clsListener::onAcceptError(QAbstractSocket::SocketError socketError) { Q_UNUSED(socketError); } /** * @brief Slot to receive data in from client */ void clsListener::onDataIn() { clsListener::commonDecode(mpClient); } /** * @brief This signal is emitted after an error occurred. The socketError * parameter describes the type of error that occurred. * @param socketError is not a registered metatype, so for queued connections, * you will have to register it with Q_DECLARE_METATYPE() and * qRegisterMetaType(). */ void clsListener::onErrorOccurred(QAbstractSocket::SocketError socketError) { //Close the connection if ( mpClient != nullptr ) { mpClient->close(); mpClient = nullptr; } qdbg() << "Socket error(" << socketError << ") https://www.google.com/search?rls=en&q=socket+error+code+" << socketError << "&ie=UTF-8&oe=UTF-8" << socketError << ".php"; } /** * @brief This signal is emitted every time a new connection is available. */ void clsListener::onNewConnection() { mpClient = mpServer->nextPendingConnection(); if ( mpClient == nullptr || mpClient->isOpen() != true ) { return; } //Connect signals and slots QObject::connect(mpClient, &QAbstractSocket::disconnected ,mpClient, &QObject::deleteLater); QObject::connect(mpClient, &QTcpSocket::readyRead ,this, &clsListener::onDataIn); QObject::connect(mpClient ,QOverload<QAbstractSocket::SocketError>::of(&QAbstractSocket::error) ,this, &clsListener::onErrorOccurred); //Signal new connection emit newConnection(*mpClient); } /** * @brief clsListener::pGetListener * @param uint16Port : Port to search for listener using * @return Pointer to listener or nullptr if not found */ clsListener* clsListener::pGetListener(quint16 uint16Port) { mpListeners::iterator itrFound = clsListener::msmpListeners.find(uint16Port); if ( itrFound != clsListener::msmpListeners.end() ) { return itrFound->second; } return nullptr; }
This is a sample of the output from the process that is connecting to another:
D00000000000000000001S000000000000:XMLMPAM PID: 2000, Port: 8123 D00000000000000000002S000000000001:This module listening on port: 8124 D00000000000000000003S000000000001:mdFileIO Version: 1.00 D00000000000000000004S000000000001:QAbstractSocket::UnconnectedState D00000000000000000005S000000000002:Connecting to: 192.168.1.158:8123 D00000000000000000006S000000004473:clsModHelper::onNewConnWithSocket, QAbstractSocket::ConnectedState D00000000000000000007S000000008655:Connected to XMLMPAM../clsModHelper.cpp179 D00000000000000000008S000000008655:QAbstractSocket::ConnectedState D00000000000000000009S000000008655:onWrite, to: 192.168.1.158:8123, written: 62, data: {"module":"mdFileIO","msgID":"1","msgType":"init","port":8124} D00000000000000000010S000000008655:onWrite, to: 192.168.1.158:8123, written: 51, data: {"module":"mdFileIO","msgID":"2","msgType":"ready"} D00000000000000000011S000000010760:onWrite, to: 192.168.1.158:8123, written: 60, data: {"module":"mdFileIO","msgID":"3","msgType":"hb","port":8124} D00000000000000000012S000000012786:onWrite, to: 192.168.1.158:8123, written: 60, data: {"module":"mdFileIO","msgID":"4","msgType":"hb","port":8124} D00000000000000000013S000000014785:onWrite, to: 192.168.1.158:8123, written: 60, data: {"module":"mdFileIO","msgID":"5","msgType":"hb","port":8124} D00000000000000000014S000000016786:onWrite, to: 192.168.1.158:8123, written: 60, data: {"module":"mdFileIO","msgID":"6","msgType":"hb","port":8124} D00000000000000000015S000000018786:onWrite, to: 192.168.1.158:8123, written: 60, data: {"module":"mdFileIO","msgID":"7","msgType":"hb","port":8124} D00000000000000000016S000000020787:onWrite, to: 192.168.1.158:8123, written: 60, data: {"module":"mdFileIO","msgID":"8","msgType":"hb","port":8124} D00000000000000000017S000000022785:onWrite, to: 192.168.1.158:8123, written: 60, data: {"module":"mdFileIO","msgID":"9","msgType":"hb","port":8124} D00000000000000000018S000000024786:onWrite, to: 192.168.1.158:8123, written: 61, data: {"module":"mdFileIO","msgID":"10","msgType":"hb","port":8124}
However I'm not seeing anything received by the other application, its as if the readyRead signal isn't being triggered.
-
@SGaist, fixed, I searched online and found the solution, modified my function adding flush after successful write:
void clsMsgSender::onWrite(QJsonObject objJSON) { if ( mpsckClient->state() == QAbstractSocket::ConnectedState ) { //Associate this TCP socket with the output data stream QByteArray arybytMsg; arybytMsg = QJsonDocument(objJSON).toJson(QJsonDocument::Compact); //Write message qint64 int64Written = mpsckClient->write(arybytMsg); if ( int64Written > 0 ) { QString strMsg(arybytMsg); qdbg() << QString("onWrite, to: %1:%2, written: %3, data: %4") .arg(mpsckClient->peerAddress().toString()) .arg(mpsckClient->peerPort()) .arg(int64Written).arg(strMsg); mpsckClient->flush(); } } }
-
Hi,
You should reduce your class to the strict minimal to investigate your issue.
Your connection code is at the bottom lost after all the data parsing.
Remove all of that until you have the rest working.
One thing you should do: log the socket disconnection and other events like that so you know when that happens. Since you are using network tech, you should also use a tool like Wireshark to monitor the network activity of your application.
Also, if you want people to help you, provide a minimal compilable example that shows the behavior.
-
@SPlatten , the slot onNewConnection is getting called, the connection is:
QObject::connect(mpServer, &QTcpServer::newConnection ,this, &clsListener::onNewConnection);
But I don't get the readyRead signal:
mpClient = mpServer->nextPendingConnection(); if ( mpClient == nullptr || mpClient->state() != QAbstractSocket::ConnectedState ) { return; } QObject::connect(mpClient, &QTcpSocket::readyRead ,this, &clsListener::onDataIn, Qt::DirectConnection);
The fourth parameter doesn't make any difference with or without, still don't get the signal, but the sender is showing that the data has been successfully sent and the returned from write is the correct number of bytes, so why don't I get the data?
I've used the debugger to step over every line so I know mpClient is valid.
-
@SGaist Producing an example isn't easy, because there are two processes, this link shows how I use to have it:
https://www.bogotobogo.com/Qt/Qt5_QTcpServer_Multithreaded_Client_Server.phpBefore I was advised on here to use moveToThread instead of deriving my classes from QThread.
-
I see you are using IP for IPC, which is counterintuitive to my brains concept of "two processes need to communicate", in which case I think IPC between two "processes" running on the same host, named pipes? message queues? shared memory?, but seems you are using IP...OK! During testing you should also run both both processes on the same host and if the intention is to do IP comms then use the loopback interface to simplify the test case more. testing across the net introduced additional possible points of failure in your test.
-
One issue in your code is that you have early return statements without any debug information so in the end you might just return early for some reason and that's why you do not see anything happening.
-
@Kent-Dorfman , I had both processes originally working and communicating with each other, then after advice on this forum I re-worked the classes and instead of deriving my classes from QThread I used moveToThread, which introduced a lot of problems which I have now overcome some of the issues, but I still cannot receive the sent message on the host IP and port.
-
@SPlatten said in Two processes communicating...:
but I still cannot receive the sent message on the host IP and port.
and what does wireshark tell you? Are you sure the message is being sent? Do you know whether problem is sender or receiver?
-
@Kent-Dorfman , I've installed Wireshark and run capture, I don't see the messages that are being sent. I can see in the Application Output:
D00000000000000000009S000000000003:clsMsgSender::run(): mdFileIO, QAbstractSocket::ConnectedState D00000000000000000010S000000000003:onWrite, to: 192.168.1.158:8123, written: 62, data: {"module":"mdFileIO","msgID":"1","msgType":"init","port":8124} D00000000000000000011S000000000003:onWrite, to: 192.168.1.158:8123, written: 51, data: {"module":"mdFileIO","msgID":"2","msgType":"ready"} D00000000000000000012S000000001995:onWrite, to: 192.168.1.158:8123, written: 60, data: {"module":"mdFileIO","msgID":"3","msgType":"hb","port":8124} D00000000000000000013S000000003992:onWrite, to: 192.168.1.158:8123, written: 60, data: {"module":"mdFileIO","msgID":"4","msgType":"hb","port":8124}
The code that sends the output:
void clsMsgSender::onWrite(QJsonObject objJSON) { if ( mpsckClient->state() == QAbstractSocket::ConnectedState ) { //Associate this TCP socket with the output data stream QByteArray arybytMsg; arybytMsg = QJsonDocument(objJSON).toJson(QJsonDocument::Compact); //Write message qint64 int64Written = mpsckClient->write(arybytMsg); if ( int64Written > 0 ) { QString strMsg(arybytMsg); qdbg() << QString("onWrite, to: %1:%2, written: %3, data: %4") .arg(mpsckClient->peerAddress().toString()) .arg(mpsckClient->peerPort()) .arg(int64Written).arg(strMsg); } } }
So the bytes written is an indication returned by the write function that it was successful but I'm not seeing anything in the Wireshark captured output:
"No.","Time","Source","Destination","Protocol","Length","Info" "1","0.000000","0.0.0.0","255.255.255.255","DHCP","342","DHCP Inform - Transaction ID 0xf8801b27" "2","0.046229","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "3","0.049298","192.168.0.1","255.255.255.255","DHCP","321","DHCP Discover - Transaction ID 0x7e6" "4","1.046923","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "5","2.049593","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "6","3.048590","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "7","4.049566","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "8","5.050152","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "9","5.053241","192.168.0.1","255.255.255.255","DHCP","321","DHCP Discover - Transaction ID 0x19d0" "10","6.052616","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "11","7.051960","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "12","8.052561","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "13","8.969550","0.0.0.0","255.255.255.255","DHCP","342","DHCP Inform - Transaction ID 0xf8801b27" "14","9.053528","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "15","10.055774","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "16","10.058871","192.168.0.1","255.255.255.255","DHCP","321","DHCP Discover - Transaction ID 0x83c" "17","11.054927","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "18","12.055906","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols" "19","13.056473","Tp-LinkT_b7:17:bf","Broadcast","0x8899","60","Realtek Layer 2 Protocols"
-
@kshegunov , there is one process that it’s the server, it spawns other processes that communicate with the server using sockets.
It’s not really client / server, the process that launches the other processes can send notifications to the other processes, the other processes can sent multiple responses in response to a single notification.
-
Please explain what these processes are.
Seems you are writing quite a complexe client server over TCP/IP system for, in the end, stuff that is running within a single machine.
-
@SGaist , I am developing a system where each process will provide expandable functionality, the interface between the main application process and the modules will be a defined JSON system. So far I have written a module that provide's file I/O functions.
I will announce a much more detail when I have it closer to completion and publish all the information.
-
Some things you need to clarify anyway:
-
What do you mean by process ? Actual separate executables that you will start using for example QProcess ? Or separate threads that will do long term operations ?
-
What is the amount of data you expect to move around between your "central application" and your "processes" ?
We won't be able to provide useful answers if you do not share details about your current architecture. As I wrote before you seem to be implemented something very complicated which might in the end be a cannon to shoot a fly.
-
-
@SGaist , when I say a process I mean a completely separate application, not a thread in the same process.
The amount of data at the moment is not identified, but if it’s large I will split it into a multi-part message.
I am implementing something that I hope will revolutionise application development. I’ve been coding professionally since 1987 on various platforms in various languages. I’ve only been using Qt since 2016.
What I’m currently seeing is the call seems to be successful as it returns the correct number of bytes written, but using wireshark I can see no packet on the network matching the written data.
-
@SGaist, fixed, I searched online and found the solution, modified my function adding flush after successful write:
void clsMsgSender::onWrite(QJsonObject objJSON) { if ( mpsckClient->state() == QAbstractSocket::ConnectedState ) { //Associate this TCP socket with the output data stream QByteArray arybytMsg; arybytMsg = QJsonDocument(objJSON).toJson(QJsonDocument::Compact); //Write message qint64 int64Written = mpsckClient->write(arybytMsg); if ( int64Written > 0 ) { QString strMsg(arybytMsg); qdbg() << QString("onWrite, to: %1:%2, written: %3, data: %4") .arg(mpsckClient->peerAddress().toString()) .arg(mpsckClient->peerPort()) .arg(int64Written).arg(strMsg); mpsckClient->flush(); } } }