Skip to content

Commit

Permalink
Benchmark message reading/waiting/processing time per endpoint
Browse files Browse the repository at this point in the history
  • Loading branch information
Al2Klimov committed Dec 11, 2024
1 parent 50c0e9e commit 7098975
Show file tree
Hide file tree
Showing 2 changed files with 32 additions and 0 deletions.
13 changes: 13 additions & 0 deletions lib/remote/endpoint.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

#include "remote/i2-remote.hpp"
#include "remote/endpoint-ti.hpp"
#include "base/benchmark.hpp"
#include "base/ringbuffer.hpp"
#include <set>

Expand Down Expand Up @@ -43,6 +44,14 @@ class Endpoint final : public ObjectImpl<Endpoint>
void AddMessageSent(int bytes);
void AddMessageReceived(int bytes);

template<class R, class S, class P>
void AddInputTimes(const R& readTime, const S& semaphoreTime, const P& processTime)
{
m_InputReadTime += readTime;
m_InputSemaphoreTime += semaphoreTime;
m_InputProcessTime += processTime;
}

double GetMessagesSentPerSecond() const override;
double GetMessagesReceivedPerSecond() const override;

Expand All @@ -61,6 +70,10 @@ 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_InputSemaphoreTime;
Benchmark m_InputProcessTime;
};

}
Expand Down
19 changes: 19 additions & 0 deletions lib/remote/jsonrpcconnection.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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) {
Expand All @@ -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());
Expand All @@ -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, cpuBoundDuration, processingStarted);
}

l_TaskStats.InsertValue(Utility::GetTime(), 1);

auto total = ch::steady_clock::now() - start;
Expand Down

0 comments on commit 7098975

Please sign in to comment.