Merge branch 'master' into balder/use-2-transport-threads-for-proton-rpc

3 files changed, 26 insertions, 30 deletions

diff --git a/searchcore/src/apps/vespa-proton-cmd/vespa-proton-cmd.cpp b/searchcore/src/apps/vespa-proton-cmd/vespa-proton-cmd.cpp
index 416a25cfb7b..2e097e7141f 100644
--- a/searchcore/src/apps/vespa-proton-cmd/vespa-proton-cmd.cpp
+++ b/searchcore/src/apps/vespa-proton-cmd/vespa-proton-cmd.cpp
@@ -20,19 +20,19 @@ private:
     App(const App &);
     App& operator=(const App &);
 
-    FRT_Supervisor *_supervisor;
+    std::unique_ptr<fnet::frt::StandaloneFRT> _frt;
     FRT_Target     *_target;
     FRT_RPCRequest *_req;
 
 public:
-    App() : _supervisor(NULL),
-            _target(NULL),
-            _req(NULL) {}
+    App() : _frt(),
+            _target(nullptr),
+            _req(nullptr) {}
     virtual ~App()
     {
-        assert(_supervisor == NULL);
-        assert(_target == NULL);
-        assert(_req == NULL);
+        assert(!_frt);
+        assert(_target == nullptr);
+        assert(_req == nullptr);
     }
 
     int usage()
@@ -49,14 +49,13 @@ public:
 
     void initRPC()
     {
-        _supervisor = new FRT_Supervisor();
-        _req = _supervisor->AllocRPCRequest();
-        _supervisor->Start();
+        _frt = std::make_unique<fnet::frt::StandaloneFRT>();
+        _req = _frt->supervisor().AllocRPCRequest();
     }
 
     void invokeRPC(bool print, double timeout=5.0)
     {
-        if (_req == NULL)
+        if (_req == nullptr)
             return;
 
         _target->InvokeSync(_req, timeout);
@@ -66,18 +65,16 @@ public:
 
     void finiRPC()
     {
-        if (_req != NULL) {
+        if (_req != nullptr) {
             _req->SubRef();
-            _req = NULL;
+            _req = nullptr;
         }
-        if (_target != NULL) {
+        if (_target != nullptr) {
             _target->SubRef();
-            _target = NULL;
+            _target = nullptr;
         }
-        if (_supervisor != NULL) {
-            _supervisor->ShutDown(true);
-            delete _supervisor;
-            _supervisor = NULL;
+        if (_frt) {
+            _frt.reset();
         }
     }
 
@@ -115,7 +112,7 @@ public:
 
         try {
             slobrok::ConfiguratorFactory sbcfg("admin/slobrok.0");
-            slobrok::api::MirrorAPI sbmirror(*_supervisor, sbcfg);
+            slobrok::api::MirrorAPI sbmirror(_frt->supervisor(), sbcfg);
             for (int timeout = 1; timeout < 20; timeout++) {
                 if (!sbmirror.ready()) {
                     FastOS_Thread::Sleep(50*timeout);
@@ -167,7 +164,7 @@ public:
 
         try {
             slobrok::ConfiguratorFactory sbcfg("admin/slobrok.0");
-            slobrok::api::MirrorAPI sbmirror(*_supervisor, sbcfg);
+            slobrok::api::MirrorAPI sbmirror(_frt->supervisor(), sbcfg);
             for (int timeout = 1; timeout < 20; timeout++) {
                 if (!sbmirror.ready()) {
                     FastOS_Thread::Sleep(50*timeout);
@@ -249,9 +246,9 @@ public:
         }
 
         if (port != 0) {
-            _target = _supervisor->GetTarget(port);
+            _target = _frt->supervisor().GetTarget(port);
         } else {
-            _target = _supervisor->GetTarget(spec.c_str());
+            _target = _frt->supervisor().GetTarget(spec.c_str());
         }
 
         bool invoked = false;
@@ -350,7 +347,7 @@ void
 App::monitorLoop()
 {
     for (;;) {
-        FRT_RPCRequest *req = _supervisor->AllocRPCRequest();
+        FRT_RPCRequest *req = _frt->supervisor().AllocRPCRequest();
         req->SetMethodName("pandora.rtc.getIncrementalState");
         FRT_Values &params = *req->GetParams();
         params.AddInt32(2000);
@@ -365,7 +362,7 @@ App::monitorLoop()
         FRT_Value &names = rvals.GetValue(0);
         FRT_Value &values = rvals.GetValue(1);
         struct timeval tnow;
-        gettimeofday(&tnow, NULL);
+        gettimeofday(&tnow, nullptr);
 
         for (unsigned int i = 0;
              i < names._string_array._len &&
diff --git a/searchcore/src/vespa/searchcore/fdispatch/common/rpc.cpp b/searchcore/src/vespa/searchcore/fdispatch/common/rpc.cpp
index eaff3b90d78..437482dddd2 100644
--- a/searchcore/src/vespa/searchcore/fdispatch/common/rpc.cpp
+++ b/searchcore/src/vespa/searchcore/fdispatch/common/rpc.cpp
@@ -10,7 +10,7 @@ extern char FastS_VersionTag[];
 FastS_RPC::FastS_RPC(FastS_AppContext *appCtx) :
    _appCtx(appCtx),
    _transport(),
-   _supervisor(&_transport, _appCtx->GetThreadPool()),
+   _supervisor(&_transport),
    _sbregister(_supervisor, slobrok::ConfiguratorFactory("admin/slobrok.0"))
 {
 }
diff --git a/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.cpp b/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.cpp
index d2b50afb932..99cc62ce30c 100644
--- a/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/rpc_hooks.cpp
@@ -8,7 +8,6 @@
 #include <vespa/fnet/frt/supervisor.h>
 #include <vespa/fnet/transport.h>
 
-
 #include <vespa/log/log.h>
 LOG_SETUP(".proton.server.rtchooks");
 
@@ -196,7 +195,7 @@ RPCHooksBase::RPCHooksBase(Params &params)
     : _proton(params.proton),
       _docsumByRPC(std::make_unique<DocsumByRPC>(_proton.getDocsumBySlime())),
       _transport(std::make_unique<FNET_Transport>(2)),
-      _orb(std::make_unique<FRT_Supervisor>(_transport.get(), &_proton.getThreadPool())),
+      _orb(std::make_unique<FRT_Supervisor>(_transport.get())),
       _proto_rpc_adapter(std::make_unique<ProtoRpcAdapter>(
                       _proton.get_search_server(),
                       _proton.get_docsum_server(),
@@ -213,7 +212,7 @@ RPCHooksBase::open(Params & params)
     initRPC();
     _regAPI.registerName((params.identity + "/realtimecontroller").c_str());
     _orb->Listen(params.rtcPort);
-    _orb->Start();
+    _transport->Start(&_proton.getThreadPool());
     LOG(debug, "started monitoring interface");
 }
 
@@ -223,7 +222,7 @@ void
 RPCHooksBase::close()
 {
     LOG(info, "shutting down monitoring interface");
-    _orb->ShutDown(true);
+    _transport->ShutDown(true);
     _executor.shutdown();
     {
         std::lock_guard<std::mutex> guard(_stateLock);