diff options
-rw-r--r-- | messagebus/src/vespa/messagebus/network/rpcnetwork.cpp | 17 | ||||
-rw-r--r-- | messagebus/src/vespa/messagebus/network/rpcnetwork.h | 8 |
2 files changed, 7 insertions, 18 deletions
diff --git a/messagebus/src/vespa/messagebus/network/rpcnetwork.cpp b/messagebus/src/vespa/messagebus/network/rpcnetwork.cpp index bbc9fc94d14..e049e436d03 100644 --- a/messagebus/src/vespa/messagebus/network/rpcnetwork.cpp +++ b/messagebus/src/vespa/messagebus/network/rpcnetwork.cpp @@ -81,7 +81,7 @@ struct TargetPoolTask : public FNET_Task { }; std::unique_ptr<vespalib::SyncableThreadExecutor> -createSingleExecutor(RPCNetworkParams::OptimizeFor optimizeFor) { +createExecutor(RPCNetworkParams::OptimizeFor optimizeFor) { switch (optimizeFor) { case RPCNetworkParams::OptimizeFor::LATENCY: return std::make_unique<vespalib::ThreadStackExecutor>(1, 0x10000); @@ -91,14 +91,6 @@ createSingleExecutor(RPCNetworkParams::OptimizeFor optimizeFor) { } } -std::unique_ptr<vespalib::SyncableThreadExecutor> -createExecutor(RPCNetworkParams::OptimizeFor optimizeFor, uint32_t numThreads) { - if ((optimizeFor == RPCNetworkParams::OptimizeFor::LATENCY) || (numThreads >= 2)) { - return std::make_unique<vespalib::ThreadStackExecutor>(numThreads, 0x10000); - } - return std::make_unique<vespalib::SingleExecutor>(1000, 10, 1ms); -} - } RPCNetwork::SendContext::SendContext(RPCNetwork &net, const Message &msg, @@ -155,10 +147,9 @@ RPCNetwork::RPCNetwork(const RPCNetworkParams ¶ms) : _targetPool(std::make_unique<RPCTargetPool>(params.getConnectionExpireSecs())), _targetPoolTask(std::make_unique<TargetPoolTask>(_scheduler, *_targetPool)), _servicePool(std::make_unique<RPCServicePool>(*_mirror, 4096)), - _singleEncodeExecutor(createSingleExecutor(params.getOptimizeFor())), - _singleDecodeExecutor(createSingleExecutor(params.getOptimizeFor())), - _encodeExecutor(createExecutor(params.getOptimizeFor(), std::max(1u, params.getNumThreads()/4))), - _decodeExecutor(createExecutor(params.getOptimizeFor(), std::max(1u, params.getNumThreads()/4))), + _singleEncodeExecutor(createExecutor(params.getOptimizeFor())), + _singleDecodeExecutor(createExecutor(params.getOptimizeFor())), + _executor(std::make_unique<vespalib::ThreadStackExecutor>(params.getNumThreads(), 128000)), _sendV1(std::make_unique<RPCSendV1>()), _sendV2(std::make_unique<RPCSendV2>()), _sendAdapters(), diff --git a/messagebus/src/vespa/messagebus/network/rpcnetwork.h b/messagebus/src/vespa/messagebus/network/rpcnetwork.h index 625efd7ae27..a37cf1d9176 100644 --- a/messagebus/src/vespa/messagebus/network/rpcnetwork.h +++ b/messagebus/src/vespa/messagebus/network/rpcnetwork.h @@ -65,11 +65,9 @@ private: std::unique_ptr<RPCTargetPool> _targetPool; std::unique_ptr<FNET_Task> _targetPoolTask; std::unique_ptr<RPCServicePool> _servicePool; - // TODO Instead of having 4 different executors, we should move the SequencedTaskExecutor into vespalib and use it there. std::unique_ptr<vespalib::SyncableThreadExecutor> _singleEncodeExecutor; std::unique_ptr<vespalib::SyncableThreadExecutor> _singleDecodeExecutor; - std::unique_ptr<vespalib::SyncableThreadExecutor> _encodeExecutor; - std::unique_ptr<vespalib::SyncableThreadExecutor> _decodeExecutor; + std::unique_ptr<vespalib::SyncableThreadExecutor> _executor; std::unique_ptr<RPCSendAdapter> _sendV1; std::unique_ptr<RPCSendAdapter> _sendV2; SendAdapterMap _sendAdapters; @@ -227,8 +225,8 @@ public: const slobrok::api::IMirrorAPI &getMirror() const override; CompressionConfig getCompressionConfig() { return _compressionConfig; } void invoke(FRT_RPCRequest *req); - vespalib::Executor & getEncodeExecutor(bool requireSequencing) const { return requireSequencing ? *_singleEncodeExecutor : *_encodeExecutor; } - vespalib::Executor & getDecodeExecutor(bool requireSequencing) const { return requireSequencing ? *_singleDecodeExecutor : *_decodeExecutor; } + vespalib::Executor & getEncodeExecutor(bool requireSequencing) const { return requireSequencing ? *_singleEncodeExecutor : *_executor; } + vespalib::Executor & getDecodeExecutor(bool requireSequencing) const { return requireSequencing ? *_singleDecodeExecutor : *_executor; } bool allowDispatchForEncode() const { return _allowDispatchForEncode; } bool allowDispatchForDecode() const { return _allowDispatchForDecode; } |