Control events-before-wakeup for mbus too.

3 files changed, 10 insertions, 31 deletions

diff --git a/messagebus/src/vespa/messagebus/network/rpcnetwork.cpp b/messagebus/src/vespa/messagebus/network/rpcnetwork.cpp
index 77d8ca24cfc..8f9037f70dd 100644
--- a/messagebus/src/vespa/messagebus/network/rpcnetwork.cpp
+++ b/messagebus/src/vespa/messagebus/network/rpcnetwork.cpp
@@ -84,7 +84,8 @@ toFNETConfig(const RPCNetworkParams & params) {
     return fnet::TransportConfig(params.getNumNetworkThreads())
               .maxInputBufferSize(params.getMaxInputBufferSize())
               .maxOutputBufferSize(params.getMaxOutputBufferSize())
-              .tcpNoDelay(params.getTcpNoDelay());
+              .tcpNoDelay(params.getTcpNoDelay())
+              .events_before_wakeup(params.events_before_wakeup());
 }
 
 }
diff --git a/messagebus/src/vespa/messagebus/network/rpcnetworkparams.cpp b/messagebus/src/vespa/messagebus/network/rpcnetworkparams.cpp
index a4cdd25040b..8fcadaa64c6 100644
--- a/messagebus/src/vespa/messagebus/network/rpcnetworkparams.cpp
+++ b/messagebus/src/vespa/messagebus/network/rpcnetworkparams.cpp
@@ -18,11 +18,10 @@ RPCNetworkParams::RPCNetworkParams(config::ConfigUri configUri) :
     _numThreads(4),
     _numNetworkThreads(1),
     _numRpcTargets(1),
+    _events_before_wakeup(1),
     _tcpNoDelay(true),
     _dispatchOnEncode(true),
     _dispatchOnDecode(false),
-    _skip_request_thread(false),
-    _skip_reply_thread(false),
     _connectionExpireSecs(600),
     _compressionConfig(CompressionConfig::LZ4, 6, 90, 1024)
 { }
diff --git a/messagebus/src/vespa/messagebus/network/rpcnetworkparams.h b/messagebus/src/vespa/messagebus/network/rpcnetworkparams.h
index 16b8b9b1570..a8d611df653 100644
--- a/messagebus/src/vespa/messagebus/network/rpcnetworkparams.h
+++ b/messagebus/src/vespa/messagebus/network/rpcnetworkparams.h
@@ -22,11 +22,10 @@ private:
     uint32_t          _numThreads;
     uint32_t          _numNetworkThreads;
     uint32_t          _numRpcTargets;
+    uint32_t          _events_before_wakeup;
     bool              _tcpNoDelay;
     bool              _dispatchOnEncode;
     bool              _dispatchOnDecode;
-    bool              _skip_request_thread;
-    bool              _skip_reply_thread;
     double            _connectionExpireSecs;
     CompressionConfig _compressionConfig;
 
@@ -164,19 +163,6 @@ public:
     }
 
     /**
-     * Sets the maximum input buffer size allowed for the underlying FNET connection. Using the value 0 means that there
-     * is no limit; the connection will not free any allocated memory until it is cleaned up. This might potentially
-     * save alot of allocation time.
-     *
-     * @param maxInputBufferSize The maximum number of bytes.
-     * @return This, to allow chaining.
-     */
-    RPCNetworkParams &setMaxInputBufferSize(uint32_t maxInputBufferSize) {
-        _maxInputBufferSize = maxInputBufferSize;
-        return *this;
-    }
-
-    /**
      * Returns the maximum output buffer size allowed for the underlying FNET connection.
      *
      * @return The maximum number of bytes.
@@ -185,19 +171,6 @@ public:
         return _maxOutputBufferSize;
     }
 
-    /**
-     * Sets the maximum output buffer size allowed for the underlying FNET connection. Using the value 0 means that there
-     * is no limit; the connection will not free any allocated memory until it is cleaned up. This might potentially
-     * save alot of allocation time.
-     *
-     * @param maxOutputBufferSize The maximum number of bytes.
-     * @return This, to allow chaining.
-     */
-    RPCNetworkParams &setMaxOutputBufferSize(uint32_t maxOutputBufferSize) {
-        _maxOutputBufferSize = maxOutputBufferSize;
-        return *this;
-    }
-
     RPCNetworkParams &setCompressionConfig(CompressionConfig compressionConfig) {
         _compressionConfig = compressionConfig;
         return *this;
@@ -218,6 +191,12 @@ public:
     }
 
     bool getDispatchOnEncode() const { return _dispatchOnEncode; }
+
+    RPCNetworkParams &events_before_wakeup(uint32_t value) {
+        _events_before_wakeup = value;
+        return *this;
+    }
+    uint32_t events_before_wakeup() const { return _events_before_wakeup; }
 };
 
 }