Commit 647b10fd authored by captainwong's avatar captainwong

update

parent a197448c
...@@ -32,10 +32,6 @@ ...@@ -32,10 +32,6 @@
#include <event2/bufferevent.h> #include <event2/bufferevent.h>
#include <event2/thread.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."
#endif
#define DISABLE_JLIB_LOG2 #define DISABLE_JLIB_LOG2
#include <ademco_packet.h> #include <ademco_packet.h>
...@@ -64,6 +60,7 @@ int64_t totalDecodeTime = 0; ...@@ -64,6 +60,7 @@ int64_t totalDecodeTime = 0;
struct Session { struct Session {
int fd = 0; int fd = 0;
int thread_id = 0;
bufferevent* bev = nullptr; bufferevent* bev = nullptr;
int id = 0; int id = 0;
std::string acct = {}; std::string acct = {};
...@@ -81,6 +78,19 @@ struct Session { ...@@ -81,6 +78,19 @@ struct Session {
int64_t decodeTime = 0; int64_t decodeTime = 0;
}; };
struct ThreadContext {
int thread_id = 0;
event_base* base = nullptr;
sockaddr_in addr = {};
int session_start = 0;
int session_end = 0;
void connectNext();
};
typedef ThreadContext* ThreadContextPtr;
ThreadContextPtr* threadContexts = nullptr;
void handle_ademco_msg(Session* session, bufferevent* bev) void handle_ademco_msg(Session* session, bufferevent* bev)
{ {
auto output = bufferevent_get_output(bev); auto output = bufferevent_get_output(bev);
...@@ -173,9 +183,8 @@ void eventcb(struct bufferevent* bev, short events, void* user_data) ...@@ -173,9 +183,8 @@ void eventcb(struct bufferevent* bev, short events, void* user_data)
auto session = (Session*)user_data; auto session = (Session*)user_data;
printf("eventcb on session #%d events=%04X\n", session->id, events); printf("eventcb on session #%d events=%04X\n", session->id, events);
if (events & BEV_EVENT_CONNECTED) { if (events & BEV_EVENT_CONNECTED) {
int fd = bufferevent_getfd(bev); auto ctx = threadContexts[session->thread_id];
session->fd = fd; printf("session #%d connected fd=%d\n", session->id, session->fd);
printf("session #%d connected fd=%d\n", session->id, fd);
{ {
std::lock_guard<std::mutex> lg(mutex); std::lock_guard<std::mutex> lg(mutex);
printf("live connections %d\n", ++session_connected); printf("live connections %d\n", ++session_connected);
...@@ -195,6 +204,11 @@ void eventcb(struct bufferevent* bev, short events, void* user_data) ...@@ -195,6 +204,11 @@ void eventcb(struct bufferevent* bev, short events, void* user_data)
} }
struct timeval tv = { timeout, 0 }; struct timeval tv = { timeout, 0 };
event_add(timer, &tv); event_add(timer, &tv);
if (++ctx->session_start < ctx->session_end) {
ctx->connectNext();
}
return; return;
} else if (events & (BEV_EVENT_EOF)) { } else if (events & (BEV_EVENT_EOF)) {
printf("session #%d closed.\n", session->id); printf("session #%d closed.\n", session->id);
...@@ -236,16 +250,23 @@ void eventcb(struct bufferevent* bev, short events, void* user_data) ...@@ -236,16 +250,23 @@ void eventcb(struct bufferevent* bev, short events, void* user_data)
//event_base_loopbreak(bufferevent_get_base(bev)); //event_base_loopbreak(bufferevent_get_base(bev));
} }
event_base* init_thread(const sockaddr_in& sin, int session_start, int session_per_thread) ThreadContext* init_thread(int thread_id, const sockaddr_in& sin, int session_start, int session_per_thread)
{ {
auto base = event_base_new(); ThreadContext* context = new ThreadContext();
if (!base) { context->thread_id = thread_id;
context->base = event_base_new();
if (!context->base) {
fprintf(stderr, "init libevent failed\n"); fprintf(stderr, "init libevent failed\n");
exit(-1); exit(-1);
} }
memcpy(&context->addr, &sin, sizeof(sin));
context->session_start = session_start;
context->session_end = session_start + session_per_thread;
context->connectNext();
for (int i = 0; i < session_per_thread; i++) {
auto bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE); /*for (int i = 0; i < session_per_thread; i++) {
auto bev = bufferevent_socket_new(context->base, -1, BEV_OPT_CLOSE_ON_FREE);
if (!bev) { if (!bev) {
fprintf(stderr, "allocate bufferevent failed\n"); fprintf(stderr, "allocate bufferevent failed\n");
exit(-1); exit(-1);
...@@ -265,9 +286,33 @@ event_base* init_thread(const sockaddr_in& sin, int session_start, int session_p ...@@ -265,9 +286,33 @@ event_base* init_thread(const sockaddr_in& sin, int session_start, int session_p
} }
std::this_thread::sleep_for(std::chrono::milliseconds(1)); std::this_thread::sleep_for(std::chrono::milliseconds(1));
} }*/
threadContexts[thread_id] = context;
return context;
}
return base; void ThreadContext::connectNext()
{
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->thread_id = thread_id;
session->bev = bev;
session->id = session_start;
session->acct = std::string("861234567890") + std::to_string(session_start);
session->ademco_id = 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*)(&addr), sizeof(addr)) < 0) {
fprintf(stderr, "error starting connection\n");
exit(-1);
}
session->fd = (int)bufferevent_getfd(bev);
} }
int main(int argc, char** argv) int main(int argc, char** argv)
...@@ -333,17 +378,18 @@ int main(int argc, char** argv) ...@@ -333,17 +378,18 @@ int main(int argc, char** argv)
sin.sin_port = htons(port); sin.sin_port = htons(port);
std::vector<std::thread> threads; std::vector<std::thread> threads;
threadContexts = new ThreadContextPtr[thread_count];
for (int i = 1; i < thread_count; i++) { for (int i = 1; i < thread_count; i++) {
threads.push_back(std::thread([&sin, i, session_per_thread]() { threads.push_back(std::thread([&sin, i, session_per_thread]() {
//printf("thread %lld created\n", gettid()); //printf("thread %lld created\n", gettid());
auto base = init_thread(sin, i * session_per_thread, session_per_thread); auto context = init_thread(i, sin, i * session_per_thread, session_per_thread);
event_base_dispatch(base); event_base_dispatch(context->base);
})); }));
} }
auto main_thread_base = init_thread(sin, 0, session_per_thread); auto main_thread_context = init_thread(0, sin, 0, session_per_thread);
event_base_dispatch(main_thread_base); event_base_dispatch(main_thread_context->base);
for (auto& t : threads) { for (auto& t : threads) {
t.join(); t.join();
......
<?xml version="1.0" encoding="utf-8"?> <?xml version="1.0" encoding="utf-8"?>
<Project ToolsVersion="Current" xmlns="http://schemas.microsoft.com/developer/msbuild/2003"> <Project ToolsVersion="Current" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|x64'"> <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|x64'">
<LocalDebuggerCommandArguments>192.168.1.166 12345 6 600 50</LocalDebuggerCommandArguments> <LocalDebuggerCommandArguments>192.168.1.166 12345 6 6000 50</LocalDebuggerCommandArguments>
<DebuggerFlavor>WindowsLocalDebugger</DebuggerFlavor> <DebuggerFlavor>WindowsLocalDebugger</DebuggerFlavor>
</PropertyGroup> </PropertyGroup>
</Project> </Project>
\ No newline at end of file
...@@ -6,7 +6,7 @@ opt="-std=c++11 -O3" ...@@ -6,7 +6,7 @@ opt="-std=c++11 -O3"
g++ $opt ../bench/bench.cpp -I../../include -o bench g++ $opt ../bench/bench.cpp -I../../include -o bench
g++ $opt ../demo/demo.cpp -I../../include -o demo g++ $opt ../demo/demo.cpp -I../../include -o demo
g++ $opt ../server_demo/server_demo.cpp -I../../include -lpthread -o server_demo g++ $opt ../server_demo/server_demo.cpp -I../../include -lpthread -o server_demo
g++ -std=c++11 -O3 -DNDEBUG -g -gdwarf ../server_demo_libevent/server_demo_libevent.cpp -I../../include `pkg-config --cflags --libs libevent breakpad` -levent_core -levent_pthreads -lpthread -o server_demo_libevent g++ -std=c++11 -O3 -DNDEBUG -DENABLE_BREAKPAD=$1 -g -gdwarf ../server_demo_libevent/server_demo_libevent.cpp -I../../include `pkg-config --cflags --libs libevent breakpad` -levent_core -levent_pthreads -lpthread -lbreakpad_client -o server_demo_libevent
g++ $opt ../bench_client/bench_client.cpp -I../../include `pkg-config --cflags --libs libevent` -levent_core -levent_pthreads -lpthread -o bench_client g++ $opt ../bench_client/bench_client.cpp -I../../include `pkg-config --cflags --libs libevent` -levent_core -levent_pthreads -lpthread -o bench_client
...@@ -28,7 +28,7 @@ ...@@ -28,7 +28,7 @@
#include <event2/bufferevent.h> #include <event2/bufferevent.h>
#include <event2/thread.h> #include <event2/thread.h>
#ifdef ENABLE_BREAKPAD #if ENABLE_BREAKPAD
#ifdef _WIN32 #ifdef _WIN32
#include <client/windows/handler/exception_handler.h> #include <client/windows/handler/exception_handler.h>
static bool dumpCallback(const wchar_t* dump_path, static bool dumpCallback(const wchar_t* dump_path,
...@@ -300,7 +300,7 @@ void accept_cb(evconnlistener* listener, evutil_socket_t fd, sockaddr* addr, int ...@@ -300,7 +300,7 @@ void accept_cb(evconnlistener* listener, evutil_socket_t fd, sockaddr* addr, int
auto sin = (sockaddr_in*)addr; auto sin = (sockaddr_in*)addr;
inet_ntop(AF_INET, &sin->sin_addr, str, INET_ADDRSTRLEN); inet_ntop(AF_INET, &sin->sin_addr, str, INET_ADDRSTRLEN);
printf("accpet TCP connection #%d from: %s:%d\n", (int)fd, str, sin->sin_port); printf("accpet TCP connection #%d from: %s:%d\n", (int)fd, str, sin->sin_port);
evutil_make_socket_nonblocking(fd); //evutil_make_socket_nonblocking(fd);
static int worker_id = 0; static int worker_id = 0;
auto context = worker_thread_contexts[worker_id]; auto context = worker_thread_contexts[worker_id];
...@@ -385,18 +385,22 @@ int main(int argc, char** argv) ...@@ -385,18 +385,22 @@ int main(int argc, char** argv)
fprintf(stderr, "failed to init libevent with thread by calling evthread_use_windows_threads\n"); fprintf(stderr, "failed to init libevent with thread by calling evthread_use_windows_threads\n");
return -1; return -1;
} }
google_breakpad::ExceptionHandler eh(nullptr, // dump_path #if ENABLE_BREAKPAD
google_breakpad::ExceptionHandler eh(L"./", // dump_path
nullptr, // FilterCallback nullptr, // FilterCallback
dumpCallback, // MinidumpCallback dumpCallback, // MinidumpCallback
nullptr, // callback_context nullptr, // callback_context
google_breakpad::ExceptionHandler::HANDLER_ALL // handler_types google_breakpad::ExceptionHandler::HANDLER_ALL // handler_types
); // MINIDUMP_TYPE ); // MINIDUMP_TYPE
#endif
#else #else
if (0 != evthread_use_pthreads()) { if (0 != evthread_use_pthreads()) {
fprintf(stderr, "failed to init libevent with thread by calling evthread_use_pthreads\n"); fprintf(stderr, "failed to init libevent with thread by calling evthread_use_pthreads\n");
return -1; return -1;
} }
#if ENABLE_BREAKPAD
google_breakpad::ExceptionHandler eh(google_breakpad::MinidumpDescriptor("./"), nullptr, dumpCallback, nullptr, true, -1); google_breakpad::ExceptionHandler eh(google_breakpad::MinidumpDescriptor("./"), nullptr, dumpCallback, nullptr, true, -1);
#endif
#endif #endif
int port = 12345; int port = 12345;
......
<?xml version="1.0" encoding="utf-8"?> <?xml version="1.0" encoding="utf-8"?>
<Project ToolsVersion="Current" xmlns="http://schemas.microsoft.com/developer/msbuild/2003"> <Project ToolsVersion="Current" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|x64'"> <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|x64'">
<LocalDebuggerCommandArguments>12345 4 0</LocalDebuggerCommandArguments> <LocalDebuggerCommandArguments>12345 1 1</LocalDebuggerCommandArguments>
<DebuggerFlavor>WindowsLocalDebugger</DebuggerFlavor>
</PropertyGroup>
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|x64'">
<LocalDebuggerCommandArguments>12345 1 1</LocalDebuggerCommandArguments>
<DebuggerFlavor>WindowsLocalDebugger</DebuggerFlavor> <DebuggerFlavor>WindowsLocalDebugger</DebuggerFlavor>
</PropertyGroup> </PropertyGroup>
</Project> </Project>
\ No newline at end of file
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