From 92cd615281f2faeb86037a34cb34834618b641a6 Mon Sep 17 00:00:00 2001 From: Jonas Herzig Date: Sun, 7 Jan 2018 22:45:25 +0100 Subject: [PATCH] GRPC: use AsyncNotifyWhenDone instead of cleanup timer --- src/murmur/MurmurGRPCImpl.cpp | 112 +++++++++--------- src/murmur/MurmurGRPCImpl.h | 3 - src/murmur_grpcwrapper_protoc_plugin/main.cpp | 18 +++ 3 files changed, 72 insertions(+), 61 deletions(-) diff --git a/src/murmur/MurmurGRPCImpl.cpp b/src/murmur/MurmurGRPCImpl.cpp index 74ebfe6b8..01ad49f31 100644 --- a/src/murmur/MurmurGRPCImpl.cpp +++ b/src/murmur/MurmurGRPCImpl.cpp @@ -132,76 +132,19 @@ void GRPCStop() { } } -MurmurRPCImpl::MurmurRPCImpl(const QString &address, std::shared_ptr<::grpc::ServerCredentials> credentials) : m_cleanupTimer(this) { +MurmurRPCImpl::MurmurRPCImpl(const QString &address, std::shared_ptr<::grpc::ServerCredentials> credentials) { ::grpc::ServerBuilder builder; builder.AddListeningPort(u8(address), credentials); builder.RegisterService(&m_V1Service); m_completionQueue = builder.AddCompletionQueue(); m_server = builder.BuildAndStart(); meta->connectListener(this); - connect(&m_cleanupTimer, SIGNAL(timeout()), this, SLOT(cleanup())); - m_cleanupTimer.setSingleShot(false); - m_cleanupTimer.start(1000 * 60); start(); } MurmurRPCImpl::~MurmurRPCImpl() { } -// This function periodically runs to clean up disconnected listeners. We need -// this because (as of 2015-07-21) the grpc library does not tell us when a -// client disconnects. -void MurmurRPCImpl::cleanup() { - for (auto i = m_metaServiceListeners.begin(); i != m_metaServiceListeners.end(); ) { - auto listener = *i; - if (listener->context.IsCancelled()) { - i = m_metaServiceListeners.erase(i); - listener->deref(); - } else { - ++i; - } - } - - for (auto i = m_contextActionListeners.begin(); i != m_contextActionListeners.end(); ) { - auto &ref = i.value(); - for (auto j = ref.begin(); j != ref.end(); ) { - auto listener = j.value(); - if (listener->context.IsCancelled()) { - j = ref.erase(j); - listener->deref(); - } else { - ++j; - } - } - - if (ref.isEmpty()) { - i = m_contextActionListeners.erase(i); - } else { - ++i; - } - } - - for (auto i = m_serverServiceListeners.begin(); i != m_serverServiceListeners.end(); ) { - auto listener = i.value(); - if (listener->context.IsCancelled()) { - i = m_serverServiceListeners.erase(i); - listener->deref(); - } else { - ++i; - } - } - - for (auto i = m_authenticators.begin(); i != m_authenticators.end(); ) { - auto listener = i.value(); - if (listener->context.IsCancelled()) { - i = m_authenticators.erase(i); - listener->deref(); - } else { - ++i; - } - } -} - // ToRPC/FromRPC methods convert data to/from grpc protocol buffer messages. void ToRPC(const ::Server *srv, const ::Channel *c, ::MurmurRPC::Channel *rc) { rc->mutable_server()->set_id(srv->iServerNum); @@ -1265,6 +1208,15 @@ void V1_ServerEvents::impl(bool) { rpc->m_serverServiceListeners.insert(server->iServerNum, this); } +void V1_ServerEvents::done(bool) { + auto &ssls = rpc->m_serverServiceListeners; + auto i = std::find(ssls.begin(), ssls.end(), this); + if (i != ssls.end()) { + ssls.erase(i); + } + deref(); +} + void V1_GetUptime::impl(bool) { ::MurmurRPC::Uptime uptime; uptime.set_secs(meta->tUptime.elapsed()/1000000LL); @@ -1285,6 +1237,15 @@ void V1_Events::impl(bool) { rpc->m_metaServiceListeners.insert(this); } +void V1_Events::done(bool) { + auto &msls = rpc->m_metaServiceListeners; + auto i = std::find(msls.begin(), msls.end(), this); + if (i != msls.end()) { + msls.erase(i); + } + deref(); +} + void V1_ContextActionAdd::impl(bool) { auto server = MustServer(request); auto user = MustUser(server, request); @@ -1350,6 +1311,24 @@ void V1_ContextActionEvents::impl(bool) { rpc->m_contextActionListeners[server->iServerNum].insert(u8(request.action()), this); } +void V1_ContextActionEvents::done(bool) { + auto server = MustServer(request); + auto &cals = rpc->m_contextActionListeners; + auto &scals = cals[server->iServerNum]; + + auto i = std::find(scals.begin(), scals.end(), this); + if (i != scals.end()) { + scals.erase(i); + } + deref(); + if (scals.isEmpty()) { + auto i = std::find(cals.begin(), cals.end(), scals); + if (i != cals.end()) { + cals.erase(i); + } + } +} + void V1_TextMessageSend::impl(bool) { auto server = MustServer(request); @@ -1393,6 +1372,15 @@ void V1_TextMessageFilter::impl(bool) { stream.Read(&request, callback(onInitialize)); } +void V1_TextMessageFilter::done(bool) { + auto server = MustServer(request); + auto filter = rpc->m_textMessageFilters.value(server->iServerNum); + if (filter == this) { + rpc->m_textMessageFilters.remove(server->iServerNum); + } + deref(); +} + void V1_LogQuery::impl(bool) { auto serverID = MustServerID(request); @@ -1981,6 +1969,14 @@ void V1_AuthenticatorStream::impl(bool) { stream.Read(&request, callback(onInitialize)); } +void V1_AuthenticatorStream::done(bool) { + auto i = std::find(rpc->m_authenticators.begin(), rpc->m_authenticators.end(), this); + if (i != rpc->m_authenticators.end()) { + rpc->m_authenticators.erase(i); + } + deref(); +} + void V1_DatabaseUserQuery::impl(bool) { auto server = MustServer(request); diff --git a/src/murmur/MurmurGRPCImpl.h b/src/murmur/MurmurGRPCImpl.h index 3fee2f6ee..14aeddb04 100644 --- a/src/murmur/MurmurGRPCImpl.h +++ b/src/murmur/MurmurGRPCImpl.h @@ -36,7 +36,6 @@ namespace MurmurRPC { class MurmurRPCImpl : public QThread { Q_OBJECT; std::unique_ptr m_server; - QTimer m_cleanupTimer; protected: void customEvent(QEvent *evt); public: @@ -76,8 +75,6 @@ class MurmurRPCImpl : public QThread { void sendServerEvent(const ::Server *s, const ::MurmurRPC::Server_Event &e); public slots: - void cleanup(); - void started(Server *server); void stopped(Server *server); diff --git a/src/murmur_grpcwrapper_protoc_plugin/main.cpp b/src/murmur_grpcwrapper_protoc_plugin/main.cpp index 2d728485a..12014435d 100644 --- a/src/murmur_grpcwrapper_protoc_plugin/main.cpp +++ b/src/murmur_grpcwrapper_protoc_plugin/main.cpp @@ -54,6 +54,7 @@ public: } void impl(bool ok); + void done(bool ok); ::boost::function *callback(::boost::function cb) { auto fn = ::boost::bind(&$service$_$method$::callbackAction, this, cb, _1); @@ -66,8 +67,16 @@ public: QCoreApplication::instance()->postEvent(rpc, ie); } + void handleDone(bool ok) { + auto ie = new RPCExecEvent(::boost::bind(&$service$_$method$::done, this, ok), this); + QCoreApplication::instance()->postEvent(rpc, ie); + } + static void create(MurmurRPCImpl *rpc, ::$ns$::$service$::AsyncService *service) { auto call = new $service$_$method$(rpc, service); + auto done_fn = ::boost::bind(&$service$_$method$::handleDone, call, _1); + auto done_fn_ptr = new ::boost::function(done_fn); + call->context.AsyncNotifyWhenDone(done_fn_ptr); auto fn = ::boost::bind(&$service$_$method$::handle, call, _1); auto fn_ptr = new ::boost::function(fn); service->Request$method$(&call->context, &call->request, &call->stream, rpc->m_completionQueue.get(), rpc->m_completionQueue.get(), fn_ptr); @@ -122,6 +131,7 @@ public: } void impl(bool ok); + void done(bool ok); bool write() { bool processed = false; @@ -166,8 +176,16 @@ public: QCoreApplication::instance()->postEvent(rpc, ie); } + void handleDone(bool ok) { + auto ie = new RPCExecEvent(::boost::bind(&$service$_$method$::done, this, ok), this); + QCoreApplication::instance()->postEvent(rpc, ie); + } + static void create(MurmurRPCImpl *rpc, ::$ns$::$service$::AsyncService *service) { auto call = new $service$_$method$(rpc, service); + auto done_fn = ::boost::bind(&$service$_$method$::handleDone, call, _1); + auto done_fn_ptr = new ::boost::function(done_fn); + call->context.AsyncNotifyWhenDone(done_fn_ptr); auto fn = ::boost::bind(&$service$_$method$::handle, call, _1); auto fn_ptr = new ::boost::function(fn); service->Request$method$(&call->context, &call->stream, rpc->m_completionQueue.get(), rpc->m_completionQueue.get(), fn_ptr);