diff --git a/examples/bench_client/bench_client.cpp b/examples/bench_client/bench_client.cpp index 18f886a79c858d1a336bfd2fd420e97d826897e9..d4009a856801b68fd7c4ec964fe2e076f47efd7f 100644 --- a/examples/bench_client/bench_client.cpp +++ b/examples/bench_client/bench_client.cpp @@ -25,10 +25,13 @@ Ping Pong Client #include <string> #include <algorithm> #include <chrono> +#include <thread> +#include <mutex> #include <event2/listener.h> #include <event2/buffer.h> #include <event2/bufferevent.h> +#include <event2/thread.h> #if !defined(LIBEVENT_VERSION_NUMBER) || LIBEVENT_VERSION_NUMBER < 0x02010100 #error "This version of Libevent is not supported; Get 2.1.1-alpha or later." @@ -39,10 +42,12 @@ Ping Pong Client using namespace ademco; +int thread_count = 0; int session_count = 0; int session_connected = 0; int timeout = 0; +std::mutex mutex = {}; int64_t totalBytesRead = 0; int64_t totalBytesWrite = 0; int64_t totalMsgRead = 0; @@ -161,14 +166,15 @@ void eventcb(struct bufferevent* bev, short events, void* user_data) printf("eventcb #%d events=%04X\n", session->id, events); if (events & BEV_EVENT_CONNECTED) { printf("client #%d connected\n", session->id); - if (++session_connected == session_count) { - 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); + std::lock_guard<std::mutex> lg(mutex); + if (++session_connected == session_count) { + 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 timer = event_new(base, bufferevent_getfd(bev), EV_TIMEOUT, timer_cb, bev); if (!timer) { @@ -186,31 +192,65 @@ void eventcb(struct bufferevent* bev, short events, void* user_data) session->id, strerror(errno)); } - totalBytesRead += session->bytesRead; - totalBytesWrite += session->bytesWritten; - totalMsgRead += session->msgRead; - totalMsgWritten += session->msgWritten; - totalEncodeTime += session->encodeTime; - totalDecodeTime += session->decodeTime; + { + std::lock_guard<std::mutex> lg(mutex); + totalBytesRead += session->bytesRead; + totalBytesWrite += session->bytesWritten; + totalMsgRead += session->msgRead; + 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; + bufferevent_free(bev); +} - 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); +event_base* init_thread(const sockaddr_in& sin, int session_start, int session_per_thread) +{ + auto base = event_base_new(); + if (!base) { + fprintf(stderr, "init libevent failed\n"); + exit(-1); } - 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) @@ -218,10 +258,19 @@ int main(int argc, char** argv) #ifdef _WIN32 WSADATA 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 - if (argc < 5) { - printf("Usage: %s ip port session_count timeout\ntimeout is in seconds\n", argv[0]); + if (argc < 6) { + printf("Usage: %s ip port thread_count session_count timeout\ntimeout is in seconds\n", argv[0]); return 1; } @@ -231,54 +280,54 @@ int main(int argc, char** argv) puts("Invalid port"); 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) { puts("Invalid session_count"); 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) { puts("Invalid timeout"); return 1; } printf("using libevent %s\n", event_get_version()); - printf("starting %s to %s:%d with %d sessions, timeout=%ds\n", - argv[0], ip, port, session_count, timeout); + int session_per_thread = session_count / thread_count; + 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 }; sin.sin_family = AF_INET; sin.sin_addr.s_addr = inet_addr(ip); sin.sin_port = htons(port); - auto base = event_base_new(); - if (!base) { - fprintf(stderr, "init libevent failed\n"); - return -1; + std::vector<std::thread> threads; + + for (int i = 1; i < thread_count; i++) { + 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 bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE); - 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); + auto main_thread_base = init_thread(sin, 0, session_per_thread); + event_base_dispatch(main_thread_base); - if (bufferevent_socket_connect(bev, (sockaddr*)(&sin), sizeof(sin)) < 0) { - fprintf(stderr, "error starting connection\n"); - return -1; - } + for (int i = 1; i < thread_count; i++) { + threads[i].join(); } - event_base_dispatch(base); - return 0; }