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

Converted some static vars in RpcConnection:on_clientSock_dataAvail() into member vars to avoid contention across connections; fixed condition which causes a portgroup to never query for stats (it assumes a GetStats is already outstanding) caused by a timing condition (repeated connect/disconnect); improved debug output for RPCs on both client and server side

Browse Source
This commit is contained in:
Srivats P. 2014-05-19 18:26:20 +05:30
parent b470ed84bb
commit f7c3c06845
5 changed files with 67 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
client/portgroup.cpp
View File

@ -138,6 +138,8 @@ void PortGroup::on_rpcChannel_disconnected()
emit portListChanged(mPortGroupId); emit portListChanged(mPortGroupId);
emit portGroupDataChanged(mPortGroupId); emit portGroupDataChanged(mPortGroupId);
isGetStatsPending_ = false;
if (reconnect) if (reconnect)
{ {
qDebug("starting reconnect timer for %d ms ...", reconnectAfter); qDebug("starting reconnect timer for %d ms ...", reconnectAfter);

39
rpc/pbrpcchannel.cpp
View File

@ -102,9 +102,10 @@ void PbRpcChannel::CallMethod(
if (isPending) if (isPending)
{ {
RpcCall call; RpcCall call;
qDebug("RpcChannel: queueing method %d since %d is pending; " qDebug("RpcChannel: queueing rpc since method %d is pending;<----\n "
"queued message = <%s>", "queued method = %d\n"
method->index(), pendingMethodId, req->DebugString().c_str()); "queued message = \n%s\n---->",
pendingMethodId, method->index(), req->DebugString().c_str());
call.method = method; call.method = method;
call.controller = controller; call.controller = controller;
@ -122,10 +123,9 @@ void PbRpcChannel::CallMethod(
if (!req->IsInitialized()) if (!req->IsInitialized())
{ {
qWarning("RpcChannel: missing required fields in request"); qWarning("RpcChannel: missing required fields in request <----");
qDebug("%s", req->InitializationErrorString().c_str()); qDebug("req = \n%s", req->DebugString().c_str());
qDebug("error = \n%s\n--->", req->InitializationErrorString().c_str());
qFatal("exiting");
controller->SetFailed("Required fields missing"); controller->SetFailed("Required fields missing");
done->Run(); done->Run();
@ -146,9 +146,11 @@ void PbRpcChannel::CallMethod(
// Avoid printing stats since it happens every couple of seconds // Avoid printing stats since it happens every couple of seconds
if (pendingMethodId != 13) if (pendingMethodId != 13)
{ {
qDebug("client(%s) sending %d bytes encoding <%s>", __FUNCTION__, qDebug("client(%s) sending %d bytes <----", __FUNCTION__,
PB_HDR_SIZE + len, req->DebugString().c_str()); PB_HDR_SIZE + len);
BUFDUMP(msg, PB_HDR_SIZE); BUFDUMP(msg, PB_HDR_SIZE);
qDebug("method = %d\n req = \n%s\n---->",
method->index(), req->DebugString().c_str());
} }
mpSocket->write(msg, PB_HDR_SIZE); mpSocket->write(msg, PB_HDR_SIZE);
@ -256,14 +258,17 @@ void PbRpcChannel::on_mpSocket_readyRead()
// Avoid printing stats // Avoid printing stats
if (method != 13) if (method != 13)
{ {
qDebug("client(%s): Parsed as %s", __FUNCTION__, qDebug("client(%s): Received Msg <---- ", __FUNCTION__);
response->DebugString().c_str()); qDebug("method = %d\nresp = \n%s\n---->",
method, response->DebugString().c_str());
} }
if (!response->IsInitialized()) if (!response->IsInitialized())
{ {
qWarning("RpcChannel: missing required fields in response"); qWarning("RpcChannel: missing required fields in response <----");
qDebug("%s", response->InitializationErrorString().c_str()); qDebug("resp = \n%s", response->DebugString().c_str());
qDebug("error = \n%s\n--->",
response->InitializationErrorString().c_str());
controller->SetFailed("Required fields missing"); controller->SetFailed("Required fields missing");
} }
@ -286,7 +291,9 @@ void PbRpcChannel::on_mpSocket_readyRead()
if (pendingCallList.size()) if (pendingCallList.size())
{ {
RpcCall call = pendingCallList.takeFirst(); RpcCall call = pendingCallList.takeFirst();
qDebug("RpcChannel: executing queued method %d <%s>", qDebug("RpcChannel: executing queued method <----\n"
"method = %d\n"
"req = \n%s\n---->",
call.method->index(), call.request->DebugString().c_str()); call.method->index(), call.request->DebugString().c_str());
CallMethod(call.method, call.controller, call.request, call.response, CallMethod(call.method, call.controller, call.request, call.response,
call.done); call.done);
@ -298,7 +305,9 @@ _error_exit:
inStream->Skip(len); inStream->Skip(len);
_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\nreq = \n%s\n---->",
method, response->DebugString().c_str());
return; return;
} }

46
rpc/rpcconn.cpp
View File

@ -42,6 +42,8 @@ RpcConnection::RpcConnection(int socketDescriptor,
isPending = false; isPending = false;
pendingMethodId = -1; // don't care as long as isPending is false pendingMethodId = -1; // don't care as long as isPending is false
parsing = false;
} }
RpcConnection::~RpcConnection() RpcConnection::~RpcConnection()
@ -69,6 +71,7 @@ void RpcConnection::start()
qWarning("Unable to initialize TCP socket for incoming connection"); qWarning("Unable to initialize TCP socket for incoming connection");
return; return;
} }
qDebug("clientSock Thread = %p", clientSock->thread());
qDebug("accepting new connection from %s: %d", qDebug("accepting new connection from %s: %d",
clientSock->peerAddress().toString().toAscii().constData(), clientSock->peerAddress().toString().toAscii().constData(),
@ -129,8 +132,11 @@ void RpcConnection::sendRpcReply(PbRpcController *controller)
if (!response->IsInitialized()) if (!response->IsInitialized())
{ {
qWarning("response missing required fields!!"); qWarning("response missing required fields!! <----");
qDebug("%s", response->InitializationErrorString().c_str()); qDebug("response = \n%s"
"missing = \n%s---->",
response->DebugString().c_str(),
response->InitializationErrorString().c_str());
qFatal("exiting"); qFatal("exiting");
goto _exit; goto _exit;
} }
@ -144,9 +150,11 @@ void RpcConnection::sendRpcReply(PbRpcController *controller)
// Avoid printing stats since it happens once every couple of seconds // Avoid printing stats since it happens once every couple of seconds
if (pendingMethodId != 13) if (pendingMethodId != 13)
{ {
qDebug("Server(%s): sending %d bytes to client encoding <%s>", qDebug("Server(%s): sending %d bytes to client <----",
__FUNCTION__, len + PB_HDR_SIZE, response->DebugString().c_str()); __FUNCTION__, len + PB_HDR_SIZE);
//BUFDUMP(msg, len + 8); BUFDUMP(msg, len + 8);
qDebug("method = %d\nreq = \n%s---->",
pendingMethodId, response->DebugString().c_str());
} }
clientSock->write(msg, PB_HDR_SIZE); clientSock->write(msg, PB_HDR_SIZE);
@ -178,9 +186,11 @@ void RpcConnection::on_clientSock_dataAvail()
{ {
uchar msg[PB_HDR_SIZE]; uchar msg[PB_HDR_SIZE];
int msgLen; int msgLen;
#if 0
static bool parsing = false; static bool parsing = false;
static quint16 type, method; static quint16 type, method;
static quint32 len; static quint32 len;
#endif
const ::google::protobuf::MethodDescriptor *methodDesc; const ::google::protobuf::MethodDescriptor *methodDesc;
::google::protobuf::Message *req, *resp; ::google::protobuf::Message *req, *resp;
PbRpcController *controller; PbRpcController *controller;
@ -228,21 +238,35 @@ void RpcConnection::on_clientSock_dataAvail()
req = service->GetRequestPrototype(methodDesc).New(); req = service->GetRequestPrototype(methodDesc).New();
resp = service->GetResponsePrototype(methodDesc).New(); resp = service->GetResponsePrototype(methodDesc).New();
if (len) if (len) {
req->ParseFromBoundedZeroCopyStream(inStream, len); bool ok = req->ParseFromBoundedZeroCopyStream(inStream, len);
if (!ok)
qWarning("ParseFromBoundedZeroCopyStream fail "
"for method %d and len %d", method, len);
}
if (!req->IsInitialized()) if (!req->IsInitialized())
{ {
qWarning("Missing required fields in request"); qWarning("Missing required fields in request <----");
qDebug("%s", req->InitializationErrorString().c_str()); qDebug("method = %d\n"
"req = \n%s"
"missing = \n%s----->",
method, req->DebugString().c_str(),
req->InitializationErrorString().c_str());
qFatal("exiting"); qFatal("exiting");
delete req; delete req;
delete resp; delete resp;
goto _error_exit2; goto _error_exit2;
} }
//qDebug("Server(%s): successfully parsed as <%s>", __FUNCTION__,
//resp->DebugString().c_str()); if (method != 13) {
qDebug("Server(%s): successfully received/parsed msg <----", __FUNCTION__);
qDebug("method = %d\n"
"req = \n%s---->",
method,
req->DebugString().c_str());
}
controller = new PbRpcController(req, resp); controller = new PbRpcController(req, resp);

5
rpc/rpcconn.h
View File

@ -65,6 +65,11 @@ private:
bool isPending; bool isPending;
int pendingMethodId; int pendingMethodId;
bool parsing;
quint16 type;
quint16 method;
quint32 len;
}; };
#endif #endif

2
server/winpcapport.cpp
View File

@ -140,7 +140,7 @@ void WinPcapPort::PortMonitor::run()
quint64 lastTxPkts = 0; quint64 lastTxPkts = 0;
quint64 lastTxBytes = 0; quint64 lastTxBytes = 0;
qWarning("in %s", __PRETTY_FUNCTION__); qDebug("in %s", __PRETTY_FUNCTION__);
lastTs.tv_sec = 0; lastTs.tv_sec = 0;
lastTs.tv_usec = 0; lastTs.tv_usec = 0;