summaryrefslogtreecommitdiffstats
path: root/messagebus
diff options
context:
space:
mode:
authorHenning Baldersheim <balder@yahoo-inc.com>2022-06-29 07:58:08 +0000
committerHenning Baldersheim <balder@yahoo-inc.com>2022-06-29 07:58:08 +0000
commite60ef2ca182ac32903e5c2e379bf6eea9fe77a4a (patch)
tree303d9948b27baf82ead1823d0033a62c1ae66381 /messagebus
parent1dff254213b5977d1196e0160de82a582a623a72 (diff)
Always skip messenger and communicationmanager when you can.
Diffstat (limited to 'messagebus')
-rw-r--r--messagebus/src/tests/messenger/messenger.cpp2
-rw-r--r--messagebus/src/vespa/messagebus/messagebus.cpp6
-rw-r--r--messagebus/src/vespa/messagebus/messagebus.h2
-rw-r--r--messagebus/src/vespa/messagebus/messenger.cpp18
-rw-r--r--messagebus/src/vespa/messagebus/messenger.h6
-rw-r--r--messagebus/src/vespa/messagebus/network/rpcnetworkparams.h14
-rw-r--r--messagebus/src/vespa/messagebus/rpcmessagebus.cpp2
-rw-r--r--messagebus/src/vespa/messagebus/testlib/testserver.cpp2
8 files changed, 13 insertions, 39 deletions
diff --git a/messagebus/src/tests/messenger/messenger.cpp b/messagebus/src/tests/messenger/messenger.cpp
index 1ee34b386ae..acd4406a7dc 100644
--- a/messagebus/src/tests/messenger/messenger.cpp
+++ b/messagebus/src/tests/messenger/messenger.cpp
@@ -39,7 +39,7 @@ public:
TEST("messenger_test") {
- Messenger msn(true, true);
+ Messenger msn;
msn.start();
vespalib::Barrier barrier(2);
diff --git a/messagebus/src/vespa/messagebus/messagebus.cpp b/messagebus/src/vespa/messagebus/messagebus.cpp
index b4b29b84793..97be5955e9d 100644
--- a/messagebus/src/vespa/messagebus/messagebus.cpp
+++ b/messagebus/src/vespa/messagebus/messagebus.cpp
@@ -82,13 +82,13 @@ public:
namespace mbus {
-MessageBus::MessageBus(INetwork &net, ProtocolSet protocols, bool skip_request_thread, bool skip_reply_thread) :
+MessageBus::MessageBus(INetwork &net, ProtocolSet protocols) :
_network(net),
_lock(),
_routingTables(),
_sessions(),
_protocolRepository(std::make_unique<ProtocolRepository>()),
- _msn(std::make_unique<Messenger>(skip_request_thread, skip_reply_thread)),
+ _msn(std::make_unique<Messenger>()),
_resender(),
_maxPendingCount(0),
_maxPendingSize(0),
@@ -111,7 +111,7 @@ MessageBus::MessageBus(INetwork &net, const MessageBusParams &params) :
_routingTables(),
_sessions(),
_protocolRepository(std::make_unique<ProtocolRepository>()),
- _msn(std::make_unique<Messenger>(true, true)),
+ _msn(std::make_unique<Messenger>()),
_resender(),
_maxPendingCount(params.getMaxPendingCount()),
_maxPendingSize(params.getMaxPendingSize()),
diff --git a/messagebus/src/vespa/messagebus/messagebus.h b/messagebus/src/vespa/messagebus/messagebus.h
index d2e82835f3e..d270a0f3491 100644
--- a/messagebus/src/vespa/messagebus/messagebus.h
+++ b/messagebus/src/vespa/messagebus/messagebus.h
@@ -82,7 +82,7 @@ public:
* @param network The network to associate with.
* @param protocols An array of protocols to register.
*/
- MessageBus(INetwork &net, ProtocolSet protocols, bool skip_request_thread, bool skip_reply_thread);
+ MessageBus(INetwork &net, ProtocolSet protocols);
/**
* Constructs an instance of message bus. This requires a network object that it will associate with. This
diff --git a/messagebus/src/vespa/messagebus/messenger.cpp b/messagebus/src/vespa/messagebus/messenger.cpp
index 13ca5317148..1423876e95b 100644
--- a/messagebus/src/vespa/messagebus/messenger.cpp
+++ b/messagebus/src/vespa/messagebus/messenger.cpp
@@ -155,14 +155,12 @@ public:
namespace mbus {
-Messenger::Messenger(bool skip_request_thread, bool skip_reply_thread)
+Messenger::Messenger()
: _lock(),
_pool(128000),
_children(),
_queue(),
- _closed(false),
- _skip_request_thread(skip_request_thread),
- _skip_reply_thread(skip_reply_thread)
+ _closed(false)
{}
Messenger::~Messenger()
@@ -246,21 +244,13 @@ Messenger::start()
void
Messenger::deliverMessage(Message::UP msg, IMessageHandler &handler)
{
- if (_skip_request_thread) {
- handler.handleMessage(std::move(msg));
- } else {
- enqueue(std::make_unique<MessageTask>(std::move(msg), handler));
- }
+ handler.handleMessage(std::move(msg));
}
void
Messenger::deliverReply(Reply::UP reply, IReplyHandler &handler)
{
- if (_skip_reply_thread) {
- handler.handleReply(std::move(reply));
- } else {
- enqueue(std::make_unique<ReplyTask>(std::move(reply), handler));
- }
+ handler.handleReply(std::move(reply));
}
void
diff --git a/messagebus/src/vespa/messagebus/messenger.h b/messagebus/src/vespa/messagebus/messenger.h
index be4dbdb10d8..0d36e6006cb 100644
--- a/messagebus/src/vespa/messagebus/messenger.h
+++ b/messagebus/src/vespa/messagebus/messenger.h
@@ -42,17 +42,15 @@ private:
mutable std::mutex _lock;
std::condition_variable _cond;
FastOS_ThreadPool _pool;
- std::vector<ITask*> _children;
+ std::vector<ITask*> _children;
vespalib::ArrayQueue<ITask*> _queue;
bool _closed;
- const bool _skip_request_thread;
- const bool _skip_reply_thread;
protected:
void Run(FastOS_ThreadInterface *thread, void *arg) override;
public:
- Messenger(bool skip_request_thread, bool skip_reply_thread);
+ Messenger();
/**
* Frees any allocated resources. Also destroys all queued tasks.
diff --git a/messagebus/src/vespa/messagebus/network/rpcnetworkparams.h b/messagebus/src/vespa/messagebus/network/rpcnetworkparams.h
index 01834074e6f..16b8b9b1570 100644
--- a/messagebus/src/vespa/messagebus/network/rpcnetworkparams.h
+++ b/messagebus/src/vespa/messagebus/network/rpcnetworkparams.h
@@ -218,20 +218,6 @@ public:
}
bool getDispatchOnEncode() const { return _dispatchOnEncode; }
-
- RPCNetworkParams &setSkipRequestThread(bool skip_request_thread) {
- _skip_request_thread = skip_request_thread;
- return *this;
- }
-
- bool getSkipRequestThread() const { return _skip_request_thread; }
-
- RPCNetworkParams &setSkipReplyThread(bool skip_reply_thread) {
- _skip_reply_thread = skip_reply_thread;
- return *this;
- }
-
- bool getSkipReplyThread() const { return _skip_reply_thread; }
};
}
diff --git a/messagebus/src/vespa/messagebus/rpcmessagebus.cpp b/messagebus/src/vespa/messagebus/rpcmessagebus.cpp
index a79171f5af2..0e4b6673d20 100644
--- a/messagebus/src/vespa/messagebus/rpcmessagebus.cpp
+++ b/messagebus/src/vespa/messagebus/rpcmessagebus.cpp
@@ -25,7 +25,7 @@ RPCMessageBus::RPCMessageBus(const ProtocolSet &protocols,
const RPCNetworkParams &rpcParams,
const config::ConfigUri &routingCfgUri) :
_net(rpcParams),
- _bus(_net, protocols, rpcParams.getSkipRequestThread(), rpcParams.getSkipReplyThread()),
+ _bus(_net, protocols),
_agent(_bus),
_subscriber(routingCfgUri.getContext())
{
diff --git a/messagebus/src/vespa/messagebus/testlib/testserver.cpp b/messagebus/src/vespa/messagebus/testlib/testserver.cpp
index 80d9eaefe29..d289c372fda 100644
--- a/messagebus/src/vespa/messagebus/testlib/testserver.cpp
+++ b/messagebus/src/vespa/messagebus/testlib/testserver.cpp
@@ -25,7 +25,7 @@ TestServer::TestServer(const Identity &ident,
const Slobrok &slobrok,
IProtocol::SP protocol) :
net(RPCNetworkParams(slobrok.config()).setIdentity(ident)),
- mb(net, ProtocolSet().add(std::make_shared<SimpleProtocol>()).add(protocol), true, true)
+ mb(net, ProtocolSet().add(std::make_shared<SimpleProtocol>()).add(protocol))
{
mb.setupRouting(spec);
}