matthew/ostinato
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Convert RPC channel's static vars to data members

Browse Source 
Using static vars meant all connections were using the same static vars!
A bug that has existed since 2009!

This code is ugly, but since it is such a fundamental piece of code for
every operation, no refactoring is being done except to convert the
static vars to data members renaming where required to avoid name
conflicts. The one exception is that `cumLen` which was a per hdr.type
variable is now a single data member across all types. Since for a
single connection messages will be sequential this is expected to work
(fingers crossed!)

For the future we should look at replacing the custom RPC code with GRPC
(issue #305).

Fixes #304
This commit is contained in:
Srivats P 2020-05-12 21:33:54 +05:30
parent f92161e755
commit 75ce626532
2 changed files with 28 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
rpc/pbrpcchannel.cpp
View File

@ -23,8 +23,6 @@ along with this program. If not, see <http://www.gnu.org/licenses/>
#include <QtGlobal> #include <QtGlobal>
#include <qendian.h> #include <qendian.h>
static uchar msgBuf[4096];
PbRpcChannel::PbRpcChannel(QString serverName, quint16 port, PbRpcChannel::PbRpcChannel(QString serverName, quint16 port,
const ::google::protobuf::Message &notifProto) const ::google::protobuf::Message &notifProto)
: notifPrototype(notifProto) : notifPrototype(notifProto)
@ -100,7 +98,7 @@ void PbRpcChannel::CallMethod(
::google::protobuf::Message *response, ::google::protobuf::Message *response,
::google::protobuf::Closure* done) ::google::protobuf::Closure* done)
{ {
char* msg = (char*) &msgBuf[0]; char* msg = (char*) &sendBuffer_[0];
int len; int len;
bool ret; bool ret;
@ -174,9 +172,6 @@ void PbRpcChannel::on_mpSocket_readyRead()
{ {
const uchar *msg; const uchar *msg;
int msgLen; int msgLen;
static bool parsing = false;
static quint16 type, method;
static quint32 len;
_top: _top:
//qDebug("%s(entry): bytesAvail = %d", __FUNCTION__, mpSocket->bytesAvailable()); //qDebug("%s(entry): bytesAvail = %d", __FUNCTION__, mpSocket->bytesAvailable());
@ -196,7 +191,7 @@ _top:
} }
type = qFromBigEndian<quint16>(msg+0); type = qFromBigEndian<quint16>(msg+0);
method = qFromBigEndian<quint16>(msg+2); methodId = qFromBigEndian<quint16>(msg+2);
len = qFromBigEndian<quint32>(msg+4); len = qFromBigEndian<quint32>(msg+4);
if (msgLen > PB_HDR_SIZE) if (msgLen > PB_HDR_SIZE)
@ -212,7 +207,6 @@ _top:
{ {
case PB_MSG_TYPE_BINBLOB: case PB_MSG_TYPE_BINBLOB:
{ {
static quint32 cumLen = 0;
QIODevice *blob; QIODevice *blob;
int l = 0; int l = 0;
@ -251,9 +245,9 @@ _top:
goto _error_exit2; goto _error_exit2;
} }
if (pendingMethodId != method) if (pendingMethodId != methodId)
{ {
qWarning("invalid method id %d (expected = %d)", method, qWarning("invalid method id %d (expected = %d)", methodId,
pendingMethodId); pendingMethodId);
goto _error_exit2; goto _error_exit2;
} }
@ -263,8 +257,6 @@ _top:
case PB_MSG_TYPE_RESPONSE: case PB_MSG_TYPE_RESPONSE:
{ {
static quint32 cumLen = 0;
static QByteArray buffer;
int l = 0; int l = 0;
if (!isPending) if (!isPending)
@ -273,9 +265,9 @@ _top:
goto _error_exit; goto _error_exit;
} }
if (pendingMethodId != method) if (pendingMethodId != methodId)
{ {
qWarning("invalid method id %d (expected = %d)", method, qWarning("invalid method id %d (expected = %d)", methodId,
pendingMethodId); pendingMethodId);
goto _error_exit; goto _error_exit;
} }
@ -311,11 +303,11 @@ _top:
buffer.resize(0); buffer.resize(0);
// Avoid printing stats // Avoid printing stats
if (method != 13) if (methodId != 13)
{ {
qDebug("client(%s): Received Msg <---- ", __FUNCTION__); qDebug("client(%s): Received Msg <---- ", __FUNCTION__);
qDebug("method = %d:%s\nresp = %s\n%s\n---->", qDebug("method = %d:%s\nresp = %s\n%s\n---->",
method, this->method->name().c_str(), methodId, this->method->name().c_str(),
this->method->output_type()->name().c_str(), this->method->output_type()->name().c_str(),
response->DebugString().c_str()); response->DebugString().c_str());
} }
@ -335,8 +327,6 @@ _top:
} }
case PB_MSG_TYPE_ERROR: case PB_MSG_TYPE_ERROR:
{ {
static quint32 cumLen = 0;
static QByteArray error;
int l = 0; int l = 0;
msgLen = 0; msgLen = 0;
@ -348,7 +338,7 @@ _top:
} }
l = qMin(msgLen, int(len - cumLen)); l = qMin(msgLen, int(len - cumLen));
error.append(QByteArray((char*)msg, l)); errorBuf.append(QByteArray((char*)msg, l));
cumLen += l; cumLen += l;
//qDebug("%s: error rcvd %d/%d/%d", __PRETTY_FUNCTION__, l, cumLen, len); //qDebug("%s: error rcvd %d/%d/%d", __PRETTY_FUNCTION__, l, cumLen, len);
} }
@ -364,10 +354,10 @@ _top:
goto _exit; goto _exit;
static_cast<PbRpcController*>(controller)->SetFailed( static_cast<PbRpcController*>(controller)->SetFailed(
QString::fromUtf8(error, len)); QString::fromUtf8(errorBuf, len));
cumLen = 0; cumLen = 0;
error.resize(0); errorBuf.resize(0);
if (!isPending) if (!isPending)
{ {
@ -375,9 +365,9 @@ _top:
goto _error_exit2; goto _error_exit2;
} }
if (pendingMethodId != method) if (pendingMethodId != methodId)
{ {
qWarning("invalid method id %d (expected = %d)", method, qWarning("invalid method id %d (expected = %d)", methodId,
pendingMethodId); pendingMethodId);
goto _error_exit2; goto _error_exit2;
} }
@ -399,7 +389,7 @@ _top:
qDebug("client(%s): Received Notif Msg <---- ", __FUNCTION__); qDebug("client(%s): Received Notif Msg <---- ", __FUNCTION__);
qDebug("type = %d\nnotif = \n%s\n---->", qDebug("type = %d\nnotif = \n%s\n---->",
method, notif->DebugString().c_str()); methodId, notif->DebugString().c_str());
if (!notif->IsInitialized()) if (!notif->IsInitialized())
{ {
@ -409,7 +399,7 @@ _top:
notif->InitializationErrorString().c_str()); notif->InitializationErrorString().c_str());
} }
else else
emit notification(method, notif); emit notification(methodId, notif);
delete notif; delete notif;
notif = NULL; notif = NULL;
@ -460,7 +450,7 @@ _error_exit:
_error_exit2: _error_exit2:
parsing = false; parsing = false;
qDebug("client(%s) discarding received msg <----", __FUNCTION__); qDebug("client(%s) discarding received msg <----", __FUNCTION__);
qDebug("method = %d\n---->", method); qDebug("method = %d\n---->", methodId);
_exit: _exit:
// If we have some data still available continue reading/parsing // If we have some data still available continue reading/parsing
if (inStream->Next((const void**)&msg, &msgLen)) { if (inStream->Next((const void**)&msg, &msgLen)) {
@ -498,8 +488,7 @@ void PbRpcChannel::on_mpSocket_disconnected()
controller = NULL; controller = NULL;
response = NULL; response = NULL;
isPending = false; isPending = false;
// \todo convert parsing from static to data member parsing = false;
//parsing = false
pendingCallList.clear(); pendingCallList.clear();
emit disconnected(); emit disconnected();

11
rpc/pbrpcchannel.h
View File

@ -73,6 +73,17 @@ class PbRpcChannel : public QObject, public ::google::protobuf::RpcChannel
::google::protobuf::io::CopyingInputStreamAdaptor *inStream; ::google::protobuf::io::CopyingInputStreamAdaptor *inStream;
::google::protobuf::io::CopyingOutputStreamAdaptor *outStream; ::google::protobuf::io::CopyingOutputStreamAdaptor *outStream;
uchar sendBuffer_[4096];
// receive RPC related vars
bool parsing{false};
QByteArray buffer; // used for response type messages
QByteArray errorBuf; // used for error type messages
quint32 cumLen{0};
quint16 type;
quint16 methodId;
quint32 len;
public: public:
PbRpcChannel(QString serverName, quint16 port, PbRpcChannel(QString serverName, quint16 port,
const ::google::protobuf::Message &notifProto); const ::google::protobuf::Message &notifProto);