Commit 41201233 authored by captainwong's avatar captainwong

bench_client multithreaded

parent 2291a1da
...@@ -25,10 +25,13 @@ Ping Pong Client ...@@ -25,10 +25,13 @@ Ping Pong Client
#include <string> #include <string>
#include <algorithm> #include <algorithm>
#include <chrono> #include <chrono>
#include <thread>
#include <mutex>
#include <event2/listener.h> #include <event2/listener.h>
#include <event2/buffer.h> #include <event2/buffer.h>
#include <event2/bufferevent.h> #include <event2/bufferevent.h>
#include <event2/thread.h>
#if !defined(LIBEVENT_VERSION_NUMBER) || LIBEVENT_VERSION_NUMBER < 0x02010100 #if !defined(LIBEVENT_VERSION_NUMBER) || LIBEVENT_VERSION_NUMBER < 0x02010100
#error "This version of Libevent is not supported; Get 2.1.1-alpha or later." #error "This version of Libevent is not supported; Get 2.1.1-alpha or later."
...@@ -39,10 +42,12 @@ Ping Pong Client ...@@ -39,10 +42,12 @@ Ping Pong Client
using namespace ademco; using namespace ademco;
int thread_count = 0;
int session_count = 0; int session_count = 0;
int session_connected = 0; int session_connected = 0;
int timeout = 0; int timeout = 0;
std::mutex mutex = {};
int64_t totalBytesRead = 0; int64_t totalBytesRead = 0;
int64_t totalBytesWrite = 0; int64_t totalBytesWrite = 0;
int64_t totalMsgRead = 0; int64_t totalMsgRead = 0;
...@@ -161,14 +166,15 @@ void eventcb(struct bufferevent* bev, short events, void* user_data) ...@@ -161,14 +166,15 @@ void eventcb(struct bufferevent* bev, short events, void* user_data)
printf("eventcb #%d events=%04X\n", session->id, events); printf("eventcb #%d events=%04X\n", session->id, events);
if (events & BEV_EVENT_CONNECTED) { if (events & BEV_EVENT_CONNECTED) {
printf("client #%d connected\n", session->id); printf("client #%d connected\n", session->id);
if (++session_connected == session_count) {
printf("All connected\n");
}
{ {
char buf[1024]; std::lock_guard<std::mutex> lg(mutex);
session->lastTimePacketSize = session->packet.make_null(buf, sizeof(buf), session->seq, session->acct, session->ademco_id); if (++session_connected == session_count) {
evbuffer_add(bufferevent_get_output(bev), buf, session->lastTimePacketSize); printf("All connected\n");
}
} }
char buf[1024];
session->lastTimePacketSize = session->packet.make_null(buf, sizeof(buf), session->seq, session->acct, session->ademco_id);
evbuffer_add(bufferevent_get_output(bev), buf, session->lastTimePacketSize);
auto base = bufferevent_get_base(bev); auto base = bufferevent_get_base(bev);
auto timer = event_new(base, bufferevent_getfd(bev), EV_TIMEOUT, timer_cb, bev); auto timer = event_new(base, bufferevent_getfd(bev), EV_TIMEOUT, timer_cb, bev);
if (!timer) { if (!timer) {
...@@ -186,31 +192,65 @@ void eventcb(struct bufferevent* bev, short events, void* user_data) ...@@ -186,31 +192,65 @@ void eventcb(struct bufferevent* bev, short events, void* user_data)
session->id, strerror(errno)); session->id, strerror(errno));
} }
totalBytesRead += session->bytesRead; {
totalBytesWrite += session->bytesWritten; std::lock_guard<std::mutex> lg(mutex);
totalMsgRead += session->msgRead; totalBytesRead += session->bytesRead;
totalMsgWritten += session->msgWritten; totalBytesWrite += session->bytesWritten;
totalEncodeTime += session->encodeTime; totalMsgRead += session->msgRead;
totalDecodeTime += session->decodeTime; totalMsgWritten += session->msgWritten;
totalEncodeTime += session->encodeTime;
totalDecodeTime += session->decodeTime;
if (--session_connected == 0) {
printf("All disconnected\n");
double encodeSpeed = totalEncodeTime * 1.0 / totalMsgWritten;
double decodeSpeed = totalDecodeTime * 1.0 / totalMsgRead;
printf("Read %.2f MiB %lld packets, Write %.2f MiB %lld packets\n"
"Average msg size is %.2f bytes\n"
"Throughput is %.2f MiB/s %.2f packets/s\n"
"Encode average time %.2f us, %.2f packets/s\n"
"Decode average time %.2f us, %.2f packets/s\n",
totalBytesRead / 1024.0 / 1024, totalMsgRead, totalBytesWrite / 1024.0 / 1024, totalMsgWritten,
(totalBytesRead + totalBytesWrite) * 1.0 / (totalMsgRead + totalMsgWritten),
totalBytesRead * 1.0 / (timeout * 1024.0 * 1024), totalMsgRead * 1.0 / timeout,
encodeSpeed, 1000000.0 / encodeSpeed,
decodeSpeed, 1000000.0 / decodeSpeed);
}
}
delete session; delete session;
bufferevent_free(bev);
}
if (--session_connected == 0) { event_base* init_thread(const sockaddr_in& sin, int session_start, int session_per_thread)
printf("All disconnected\n"); {
double encodeSpeed = totalEncodeTime * 1.0 / totalMsgWritten; auto base = event_base_new();
double decodeSpeed = totalDecodeTime * 1.0 / totalMsgRead; if (!base) {
printf("Read %.2f MiB %lld packets, Write %.2f MiB %lld packets\n" fprintf(stderr, "init libevent failed\n");
"Average msg size is %.2f bytes\n" exit(-1);
"Throughput is %.2f MiB/s %.2f packets/s\n"
"Encode average time %.2f us, %.2f packets/s\n"
"Decode average time %.2f us, %.2f packets/s\n",
totalBytesRead / 1024.0 / 1024, totalMsgRead, totalBytesWrite / 1024.0 / 1024, totalMsgWritten,
(totalBytesRead + totalBytesWrite) * 1.0 / (totalMsgRead + totalMsgWritten),
totalBytesRead * 1.0 / (timeout * 1024.0 * 1024), totalMsgRead * 1.0 / timeout,
encodeSpeed, 1000000.0 / encodeSpeed,
decodeSpeed, 1000000.0 / decodeSpeed);
} }
bufferevent_free(bev); for (int i = 0; i < session_per_thread; i++) {
auto bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE);
if (!bev) {
fprintf(stderr, "allocate bufferevent failed\n");
exit(-1);
}
auto session = new Session();
session->id = i + session_start;
session->acct = std::string("861234567890") + std::to_string(i + session_start);
session->ademco_id = i + session_start;
session->seq = 1;
bufferevent_setcb(bev, readcb, writecb, eventcb, session);
bufferevent_enable(bev, EV_READ | EV_WRITE);
if (bufferevent_socket_connect(bev, (const sockaddr*)(&sin), sizeof(sin)) < 0) {
fprintf(stderr, "error starting connection\n");
exit(-1);
}
}
return base;
} }
int main(int argc, char** argv) int main(int argc, char** argv)
...@@ -218,10 +258,19 @@ int main(int argc, char** argv) ...@@ -218,10 +258,19 @@ int main(int argc, char** argv)
#ifdef _WIN32 #ifdef _WIN32
WSADATA wsa_data; WSADATA wsa_data;
WSAStartup(0x0201, &wsa_data); WSAStartup(0x0201, &wsa_data);
if (0 != evthread_use_windows_threads()) {
fprintf(stderr, "failed to init libevent with thread by calling evthread_use_windows_threads\n");
return -1;
}
#else
if (0 != evthread_use_pthreads()) {
fprintf(stderr, "failed to init libevent with thread by calling evthread_use_pthreads\n");
return -1;
}
#endif #endif
if (argc < 5) { if (argc < 6) {
printf("Usage: %s ip port session_count timeout\ntimeout is in seconds\n", argv[0]); printf("Usage: %s ip port thread_count session_count timeout\ntimeout is in seconds\n", argv[0]);
return 1; return 1;
} }
...@@ -231,54 +280,54 @@ int main(int argc, char** argv) ...@@ -231,54 +280,54 @@ int main(int argc, char** argv)
puts("Invalid port"); puts("Invalid port");
return 1; return 1;
} }
session_count = atoi(argv[3]); thread_count = atoi(argv[3]);
if (thread_count <= 0) {
puts("Invalid thread_count");
return 1;
}
session_count = atoi(argv[4]);
if (session_count <= 0) { if (session_count <= 0) {
puts("Invalid session_count"); puts("Invalid session_count");
return 1; return 1;
} }
if (thread_count > session_count) {
puts("thread_count must not bigger than session_count");
return 1;
}
if (session_count % thread_count) {
puts("session_count must times thread_count");
return 1;
}
timeout = atoi(argv[4]); timeout = atoi(argv[5]);
if (timeout <= 0) { if (timeout <= 0) {
puts("Invalid timeout"); puts("Invalid timeout");
return 1; return 1;
} }
printf("using libevent %s\n", event_get_version()); printf("using libevent %s\n", event_get_version());
printf("starting %s to %s:%d with %d sessions, timeout=%ds\n", int session_per_thread = session_count / thread_count;
argv[0], ip, port, session_count, timeout); printf("starting %s to %s:%d with %d threads, %d sessions, %d sessions per thread, timeout=%ds\n",
argv[0], ip, port, thread_count, session_count, session_per_thread, timeout);
sockaddr_in sin = { 0 }; sockaddr_in sin = { 0 };
sin.sin_family = AF_INET; sin.sin_family = AF_INET;
sin.sin_addr.s_addr = inet_addr(ip); sin.sin_addr.s_addr = inet_addr(ip);
sin.sin_port = htons(port); sin.sin_port = htons(port);
auto base = event_base_new(); std::vector<std::thread> threads;
if (!base) {
fprintf(stderr, "init libevent failed\n"); for (int i = 1; i < thread_count; i++) {
return -1; auto base = init_thread(sin, i * thread_count, session_per_thread);
threads.emplace_back(std::thread([&base]() { event_base_dispatch(base); }));
} }
for (int i = 0; i < session_count; i++) { auto main_thread_base = init_thread(sin, 0, session_per_thread);
auto bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE); event_base_dispatch(main_thread_base);
if (!bev) {
fprintf(stderr, "allocate bufferevent failed\n");
return 1;
}
auto session = new Session();
session->id = i;
session->acct = std::string("861234567890") + std::to_string(i);
session->ademco_id = i;
session->seq = 1;
bufferevent_setcb(bev, readcb, writecb, eventcb, session);
bufferevent_enable(bev, EV_READ | EV_WRITE);
if (bufferevent_socket_connect(bev, (sockaddr*)(&sin), sizeof(sin)) < 0) { for (int i = 1; i < thread_count; i++) {
fprintf(stderr, "error starting connection\n"); threads[i].join();
return -1;
}
} }
event_base_dispatch(base);
return 0; return 0;
} }
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment