Benchmark message reading/processing time per endpoint

Author: Al2Klimov

lib/remote/endpoint.hpp

@@ -5,6 +5,7 @@
 
 #include "remote/i2-remote.hpp"
 #include "remote/endpoint-ti.hpp"
+#include "base/benchmark.hpp"
 #include "base/ringbuffer.hpp"
 #include <set>
 
@@ -43,6 +44,13 @@ class Endpoint final : public ObjectImpl<Endpoint>
 	void AddMessageSent(int bytes);
 	void AddMessageReceived(int bytes);
 
+	template<class R, class P>
+	void AddInputTimes(const R& readTime, const P& processTime)
+	{
+		m_InputReadTime += readTime;
+		m_InputProcessTime += processTime;
+	}
+
 	double GetMessagesSentPerSecond() const override;
 	double GetMessagesReceivedPerSecond() const override;
 
@@ -61,6 +69,9 @@ class Endpoint final : public ObjectImpl<Endpoint>
 	mutable RingBuffer m_MessagesReceived{60};
 	mutable RingBuffer m_BytesSent{60};
 	mutable RingBuffer m_BytesReceived{60};
+
+	Benchmark m_InputReadTime;
+	Benchmark m_InputProcessTime;
 };
 
 }

lib/remote/jsonrpcconnection.cpp

@@ -3,6 +3,7 @@
 #include "remote/jsonrpcconnection.hpp"
 #include "remote/apilistener.hpp"
 #include "remote/apifunction.hpp"
+#include "base/benchmark.hpp"
 #include "remote/jsonrpc.hpp"
 #include "base/defer.hpp"
 #include "base/configtype.hpp"
@@ -66,11 +67,17 @@ void JsonRpcConnection::HandleIncomingMessages(boost::asio::yield_context yc)
 		return ch::duration_cast<ch::milliseconds>(d).count();
 	});
 
+	Benchmark::Clock::time_point readStarted, readFinished, processingStarted;
+
 	m_Stream->next_layer().SetSeen(&m_Seen);
 
 	while (!m_ShuttingDown) {
 		String jsonString;
 
+		if (m_Endpoint) {
+			readStarted = Benchmark::Clock::now();
+		}
+
 		try {
 			jsonString = JsonRpc::ReadMessage(m_Stream, yc, m_Endpoint ? -1 : 1024 * 1024);
 		} catch (const std::exception& ex) {
@@ -81,6 +88,10 @@ void JsonRpcConnection::HandleIncomingMessages(boost::asio::yield_context yc)
 			break;
 		}
 
+		if (m_Endpoint) {
+			readFinished = Benchmark::Clock::now();
+		}
+
 		m_Seen = Utility::GetTime();
 		if (m_Endpoint) {
 			m_Endpoint->AddMessageReceived(jsonString.GetLength());
@@ -96,13 +107,21 @@ void JsonRpcConnection::HandleIncomingMessages(boost::asio::yield_context yc)
 			// Cache the elapsed time to acquire a CPU semaphore used to detect extremely heavy workloads.
 			cpuBoundDuration = ch::steady_clock::now() - start;
 
+			if (m_Endpoint) {
+				processingStarted = Benchmark::Clock::now();
+			}
+
 			Dictionary::Ptr message = JsonRpc::DecodeMessage(jsonString);
 			if (String method = message->Get("method"); !method.IsEmpty()) {
 				rpcMethod = std::move(method);
 			}
 
 			MessageHandler(message);
 
+			if (m_Endpoint) {
+				m_Endpoint->AddInputTimes(readFinished - readStarted, processingStarted);
+			}
+
 			l_TaskStats.InsertValue(Utility::GetTime(), 1);
 
 			auto total = ch::steady_clock::now() - start;