Revert "- Create the common transport and threadpool in the main loop."

3 files changed, 22 insertions, 35 deletions

diff --git a/searchcore/src/apps/proton/proton.cpp b/searchcore/src/apps/proton/proton.cpp
index 3a31f941506..61536f924a5 100644
--- a/searchcore/src/apps/proton/proton.cpp
+++ b/searchcore/src/apps/proton/proton.cpp
@@ -5,13 +5,9 @@
 #include <vespa/metrics/metricmanager.h>
 #include <vespa/vespalib/util/signalhandler.h>
 #include <vespa/vespalib/util/programoptions.h>
-#include <vespa/vespalib/util/size_literals.h>
 #include <vespa/vespalib/io/fileutil.h>
 #include <vespa/vespalib/util/exceptions.h>
 #include <vespa/config/common/exceptions.h>
-#include <vespa/config/common/configcontext.h>
-#include <vespa/fnet/transport.h>
-#include <vespa/fastos/thread.h>
 #include <vespa/fastos/app.h>
 #include <iostream>
 #include <thread>
@@ -177,12 +173,6 @@ ExitOnSignal::operator()()
     }
 }
 
-fnet::TransportConfig
-buildTransportConfig() {
-    uint32_t numProcs = std::thread::hardware_concurrency();
-    return fnet::TransportConfig(std::max(1u, std::min(4u, numProcs/8)));
-}
-
 }
 
 int
@@ -196,14 +186,8 @@ App::Main()
         LOG(debug, "serviceidentity: '%s'", params.serviceidentity.c_str());
         LOG(debug, "subscribeTimeout: '%" PRIu64 "'", params.subscribeTimeout);
         std::chrono::milliseconds subscribeTimeout(params.subscribeTimeout);
-        FastOS_ThreadPool threadPool(128_Ki);
-
-        FNET_Transport transport(buildTransportConfig());
-        transport.Start(&threadPool);
-        config::ConfigServerSpec configServerSpec(transport);
-        config::ConfigUri identityUri(params.identity, std::make_shared<config::ConfigContext>(configServerSpec));
-        protonUP = std::make_unique<proton::Proton>(threadPool, transport, identityUri,
-                                                    _argc > 0 ? _argv[0] : "proton", subscribeTimeout);
+        config::ConfigUri identityUri(params.identity);
+        protonUP = std::make_unique<proton::Proton>(identityUri, _argc > 0 ? _argv[0] : "proton", subscribeTimeout);
         proton::Proton & proton = *protonUP;
         proton::BootstrapConfig::SP configSnapshot = proton.init();
         if (proton.hasAbortedInit()) {
@@ -244,7 +228,6 @@ App::Main()
                 EV_STOPPING("servicelayer", "clean shutdown");
             }
             protonUP.reset();
-            transport.ShutDown(true);
             EV_STOPPING("proton", "clean shutdown");
         }
     } catch (const vespalib::InvalidCommandLineArgumentsException &e) {
diff --git a/searchcore/src/vespa/searchcore/proton/server/proton.cpp b/searchcore/src/vespa/searchcore/proton/server/proton.cpp
index 8bd965497a8..0f20b0a7b47 100644
--- a/searchcore/src/vespa/searchcore/proton/server/proton.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/proton.cpp
@@ -201,8 +201,9 @@ Proton::ProtonFileHeaderContext::setClusterName(const vespalib::string & cluster
 }
 
 
-Proton::Proton(FastOS_ThreadPool & threadPool, FNET_Transport & transport, const config::ConfigUri & configUri,
-               const vespalib::string &progName, vespalib::duration subscribeTimeout)
+Proton::Proton(const config::ConfigUri & configUri,
+               const vespalib::string &progName,
+               vespalib::duration subscribeTimeout)
     : IProtonConfigurerOwner(),
       search::engine::MonitorServer(),
       IDocumentDBOwner(),
@@ -210,8 +211,8 @@ Proton::Proton(FastOS_ThreadPool & threadPool, FNET_Transport & transport, const
       IPersistenceEngineOwner(),
       ComponentConfigProducer(),
       _cpu_util(),
-      _threadPool(threadPool),
-      _transport(transport),
+      _threadPool(std::make_unique<FastOS_ThreadPool>(128_Ki)),
+      _transport(std::make_unique<FNET_Transport>(TransportConfig(1))),
       _configUri(configUri),
       _mutex(),
       _metricsHook(std::make_unique<MetricsUpdateHook>(*this)),
@@ -237,7 +238,7 @@ Proton::Proton(FastOS_ThreadPool & threadPool, FNET_Transport & transport, const
       _executor(1, 128_Ki),
       _protonDiskLayout(),
       _protonConfigurer(_executor, *this, _protonDiskLayout),
-      _protonConfigFetcher(_transport, configUri, _protonConfigurer, subscribeTimeout),
+      _protonConfigFetcher(*_transport, configUri, _protonConfigurer, subscribeTimeout),
       _shared_service(),
       _compile_cache_executor_binding(),
       _queryLimiter(),
@@ -259,10 +260,11 @@ Proton::init()
 {
     assert( ! _initStarted && ! _initComplete );
     _initStarted = true;
-    if (_threadPool.NewThread(_clock.getRunnable(), nullptr) == nullptr) {
+    _transport->Start(_threadPool.get());
+    if (_threadPool->NewThread(_clock.getRunnable(), nullptr) == nullptr) {
         throw IllegalStateException("Failed starting thread for the cheap clock");
     }
-    _protonConfigFetcher.start(_threadPool);
+    _protonConfigFetcher.start(*_threadPool);
     auto configSnapshot = _protonConfigurer.getPendingConfigSnapshot();
     assert(configSnapshot);
     auto bootstrapConfig = configSnapshot->getBootstrapConfig();
@@ -280,7 +282,7 @@ Proton::init(const BootstrapConfig::SP & configSnapshot)
 
     setBucketCheckSumType(protonConfig);
     setFS4Compression(protonConfig);
-    _shared_service = std::make_unique<SharedThreadingService>(SharedThreadingServiceConfig::make(protonConfig, hwInfo.cpu()), _transport);
+    _shared_service = std::make_unique<SharedThreadingService>(SharedThreadingServiceConfig::make(protonConfig, hwInfo.cpu()), *_transport);
     _diskMemUsageSampler = std::make_unique<DiskMemUsageSampler>(_shared_service->transport(), protonConfig.basedir,
                                                                  diskMemUsageSamplerConfig(protonConfig, hwInfo));
 
@@ -312,10 +314,10 @@ Proton::init(const BootstrapConfig::SP & configSnapshot)
         strategy = std::make_shared<SimpleFlush>();
         break;
     }
-    _protonDiskLayout = std::make_unique<ProtonDiskLayout>(_transport, protonConfig.basedir, protonConfig.tlsspec);
+    _protonDiskLayout = std::make_unique<ProtonDiskLayout>(*_transport, protonConfig.basedir, protonConfig.tlsspec);
     vespalib::chdir(protonConfig.basedir);
     vespalib::alloc::MmapFileAllocatorFactory::instance().setup(protonConfig.basedir + "/swapdirs");
-    _tls->start(_transport, hwInfo.cpu().cores());
+    _tls->start(*_transport, hwInfo.cpu().cores());
     _flushEngine = std::make_unique<FlushEngine>(std::make_shared<flushengine::TlsStatsFactory>(_tls->getTransLogServer()),
                                                  strategy, flush.maxconcurrent, vespalib::from_s(flush.idleinterval));
     _metricsEngine->addExternalMetrics(_summaryEngine->getMetrics());
@@ -477,6 +479,7 @@ Proton::~Proton()
     _compile_cache_executor_binding.reset();
     _shared_service.reset();
     _clock.stop();
+    _transport->ShutDown(true);
     LOG(debug, "Explicit destructor done");
 }
 
@@ -599,7 +602,7 @@ Proton::addDocumentDB(const document::DocumentType &docType,
 
     vespalib::string db_dir = config.basedir + "/documents/" + docTypeName.toString();
     vespalib::mkdir(db_dir, false); // Assume parent is created.
-    auto config_store = std::make_unique<FileConfigManager>(_transport, db_dir + "/config",
+    auto config_store = std::make_unique<FileConfigManager>(*_transport, db_dir + "/config",
                                                             documentDBConfig->getConfigId(), docTypeName.getName());
     config_store->setProtonConfig(bootstrapConfig->getProtonConfigSP());
     if (!initializeThreads) {
diff --git a/searchcore/src/vespa/searchcore/proton/server/proton.h b/searchcore/src/vespa/searchcore/proton/server/proton.h
index 73b8ae83ef2..0490b1e00b7 100644
--- a/searchcore/src/vespa/searchcore/proton/server/proton.h
+++ b/searchcore/src/vespa/searchcore/proton/server/proton.h
@@ -82,8 +82,8 @@ private:
     };
 
     vespalib::CpuUtil                      _cpu_util;
-    FastOS_ThreadPool                    & _threadPool;
-    FNET_Transport                       & _transport;
+    std::unique_ptr<FastOS_ThreadPool>     _threadPool;
+    std::unique_ptr<FNET_Transport>        _transport;
     const config::ConfigUri                _configUri;
     mutable std::shared_mutex              _mutex;
     std::unique_ptr<metrics::UpdateHook>   _metricsHook;
@@ -145,8 +145,9 @@ public:
     typedef std::unique_ptr<Proton> UP;
     typedef std::shared_ptr<Proton> SP;
 
-    Proton(FastOS_ThreadPool & threadPool, FNET_Transport & transport, const config::ConfigUri & configUri,
-           const vespalib::string &progName, vespalib::duration subscribeTimeout);
+    Proton(const config::ConfigUri & configUri,
+           const vespalib::string &progName,
+           vespalib::duration subscribeTimeout);
     ~Proton() override;
 
     /**
@@ -182,7 +183,7 @@ public:
                   const std::shared_ptr<DocumentDBConfig> &documentDBConfig, InitializeThreads initializeThreads);
 
     metrics::MetricManager & getMetricManager();
-    FastOS_ThreadPool & getThreadPool() { return _threadPool; }
+    FastOS_ThreadPool & getThreadPool() { return *_threadPool; }
 
     bool triggerFlush();
     bool prepareRestart();