Commit 0e418d6a authored by captainwong's avatar captainwong

server_demo_libevent multithreaded

parent 41201233
......@@ -45,13 +45,26 @@ struct Client {
uint16_t seq = 0;
};
std::unordered_map<int, Client> clients = {};
struct ThreadContext {
int worker_id = -1;
event_base* base = nullptr;
std::mutex mutex = {};
std::unordered_map<int, Client> clients = {};
AdemcoPacket packet = {};
};
event_base* listen_thread_evbase = nullptr;
int thread_count = 4;
std::vector<ThreadContext*> worker_thread_contexts = {};
// events will be sent to all clients
std::vector<ADEMCO_EVENT> events = {};
int threads_to_handled_event = 0;
std::mutex mutex = {};
char pwd[1024] = { 0 };
int disable_data_print = 0;
void op_usage()
{
printf("Press A for Arm, D for Disarm, E for Emergency, Q for Quit\n");
......@@ -59,28 +72,34 @@ void op_usage()
void usage(const char* name)
{
printf("Usage: %s listening_port\n", name);
printf("Usage: %s [listening_port] [thread_count] [disable_data_print]\n", name);
}
void handle_ademco_msg(AdemcoPacket& packet, bufferevent* bev)
void handle_ademco_msg(ThreadContext* context, bufferevent* bev)
{
int fd = (int)bufferevent_getfd(bev);
auto& client = clients[fd];
printf("C#%d acct=%s ademco_id=%06zX :%s\n", fd, client.acct.data(), client.ademco_id, packet.toString().data());
auto& client = context->clients[fd];
if (!disable_data_print) {
printf("T#%d C#%d acct=%s ademco_id=%06zX :%s\n",
context->worker_id, fd, client.acct.data(), client.ademco_id, context->packet.toString().data());
}
auto output = bufferevent_get_output(bev);
switch (packet.id_.eid_) {
switch (context->packet.id_.eid_) {
case AdemcoId::id_null:
case AdemcoId::id_hb:
case AdemcoId::id_admcid:
if (packet.id_.eid_ != AdemcoId::id_null) {
client.acct = packet.acct_.acct();
client.ademco_id = packet.ademcoData_.ademco_id_;
if (context->packet.id_.eid_ != AdemcoId::id_null) {
client.acct = context->packet.acct_.acct();
client.ademco_id = context->packet.ademcoData_.ademco_id_;
}
{
char buf[1024];
size_t n = packet.make_ack(buf, sizeof(buf), packet.seq_.value_, packet.acct_.acct(), packet.ademcoData_.ademco_id_);
size_t n = context->packet.make_ack(buf, sizeof(buf), context->packet.seq_.value_, context->packet.acct_.acct(), context->packet.ademcoData_.ademco_id_);
evbuffer_add(output, buf, n);
printf("S#%d acct=%s ademco_id=%06zX :%s\n", fd, client.acct.data(), client.ademco_id, packet.toString().data());
if (!disable_data_print) {
printf("T#%d S#%d acct=%s ademco_id=%06zX :%s\n",
context->worker_id, fd, client.acct.data(), client.ademco_id, context->packet.toString().data());
}
}
break;
default:
......@@ -89,18 +108,24 @@ void handle_ademco_msg(AdemcoPacket& packet, bufferevent* bev)
}
// send all buffered events to all clients
void commandcb(evutil_socket_t, short, void*)
void commandcb(evutil_socket_t, short, void* user_data)
{
auto context = (ThreadContext*)user_data;
std::vector<ADEMCO_EVENT> evs;
{
std::lock_guard<std::mutex> lg(mutex);
evs = std::move(events);
if (--threads_to_handled_event == 0) {
evs = std::move(events);
} else {
evs = events;
}
}
char buf[1024];
AdemcoPacket maker;
for (auto& client : clients) {
std::lock_guard<std::mutex> lg(context->mutex);
for (auto& client : context->clients) {
for (auto e : evs) {
size_t n = 0;
if (e == EVENT_DISARM) {
......@@ -108,13 +133,15 @@ void commandcb(evutil_socket_t, short, void*)
client.second.seq = 1;
}
auto xdata = makeXData(pwd, 6);
n = maker.make_hb(buf, sizeof(buf), client.second.seq, client.second.acct, client.second.ademco_id, 0, e, 0, xdata);
n = context->packet.make_hb(buf, sizeof(buf), client.second.seq, client.second.acct, client.second.ademco_id, 0, e, 0, xdata);
} else {
n = maker.make_hb(buf, sizeof(buf), client.second.seq, client.second.acct, client.second.ademco_id, 0, e, 0);
n = context->packet.make_hb(buf, sizeof(buf), client.second.seq, client.second.acct, client.second.ademco_id, 0, e, 0);
}
evbuffer_add(client.second.output, buf, n);
printf("S#%d acct=%s ademco_id=%06zX :%s\n",
client.second.fd, client.second.acct.data(), client.second.ademco_id, maker.toString().data());
if (!disable_data_print) {
printf("S#%d acct=%s ademco_id=%06zX :%s\n",
client.second.fd, client.second.acct.data(), client.second.ademco_id, context->packet.toString().data());
}
}
}
}
......@@ -122,20 +149,20 @@ void commandcb(evutil_socket_t, short, void*)
void readcb(struct bufferevent* bev, void* user_data)
{
auto input = bufferevent_get_input(bev);
auto context = (ThreadContext*)user_data;
while (1) {
size_t len = evbuffer_get_length(input);
if (len < 9) { return; }
static AdemcoPacket parser = {};
char buff[1024] = { 0 };
evbuffer_copyout(input, buff, std::min(len, sizeof(buff)));
size_t cb_commited = 0;
auto res = parser.parse(buff, 1024, cb_commited);
auto res = context->packet.parse(buff, 1024, cb_commited);
bool done = false;
switch (res) {
case ademco::ParseResult::RESULT_OK:
evbuffer_drain(input, cb_commited);
handle_ademco_msg(parser, bev);
handle_ademco_msg(context, bev);
break;
case ademco::ParseResult::RESULT_NOT_ENOUGH:
done = true;
......@@ -162,34 +189,46 @@ void eventcb(struct bufferevent* bev, short events, void* user_data)
printf("Got an error on the connection %d: %s\n",
fd, strerror(errno));
}
printf("Connection #%d closed.\n", fd);
clients.erase(fd);
printf("Connection #%d closed.\n", fd);
auto context = (ThreadContext*)user_data;
{
std::lock_guard<std::mutex> lg(context->mutex);
context->clients.erase(fd);
}
bufferevent_free(bev);
}
void accept_cb(evconnlistener* listener, evutil_socket_t fd, sockaddr* addr, int socklen, void* context)
void accept_cb(evconnlistener* listener, evutil_socket_t fd, sockaddr* addr, int socklen, void* user_data)
{
char str[INET_ADDRSTRLEN] = { 0 };
auto sin = (sockaddr_in*)addr;
inet_ntop(AF_INET, &sin->sin_addr, str, INET_ADDRSTRLEN);
printf("accpet TCP connection from: %s:%d\n", str, sin->sin_port);
//evutil_make_socket_nonblocking(fd);
auto base = evconnlistener_get_base(listener);
auto bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
evutil_make_socket_nonblocking(fd);
static int worker_id = 0;
auto context = worker_thread_contexts[worker_id];
auto bev = bufferevent_socket_new(context->base, fd, BEV_OPT_CLOSE_ON_FREE);
if (!bev) {
fprintf(stderr, "Error constructing bufferevent!\n");
event_base_loopbreak(base);
event_base_loopbreak(context->base);
return;
}
Client client;
client.fd = (int)fd;
client.output = bufferevent_get_output(bev);
clients[(int)fd] = client;
bufferevent_setcb(bev, readcb, nullptr, eventcb, nullptr);
{
std::lock_guard<std::mutex> lg(context->mutex);
context->clients[(int)fd] = client;
}
bufferevent_setcb(bev, readcb, nullptr, eventcb, context);
bufferevent_enable(bev, EV_WRITE | EV_READ);
worker_id = (++worker_id) % thread_count;
}
void accpet_error_cb(evconnlistener* listener, void* context)
......@@ -200,6 +239,47 @@ void accpet_error_cb(evconnlistener* listener, void* context)
event_base_loopexit(base, nullptr);
}
void init_listener_thread(const sockaddr_in& addr)
{
listen_thread_evbase = event_base_new();
if (!listen_thread_evbase) {
fprintf(stderr, "init libevent failed\n");
exit(-1);
}
auto listener = evconnlistener_new_bind(listen_thread_evbase,
accept_cb,
nullptr,
LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE,
-1, // backlog, -1 for default
(const sockaddr*)(&addr),
sizeof(addr));
if (!listener) {
fprintf(stderr, "create listener failed\n");
exit(-1);
}
evconnlistener_set_error_cb(listener, accpet_error_cb);
}
void dummy_cb_avoid_worker_exit(evutil_socket_t, short, void*)
{
}
ThreadContext* init_worker_thread(int i)
{
ThreadContext* context = new ThreadContext();
context->worker_id = i;
context->base = event_base_new();
if (!context->base) {
fprintf(stderr, "init libevent failed\n");
exit(-1);
}
timeval tv = { 1, 0 };
event_add(event_new(context->base, -1, EV_PERSIST, dummy_cb_avoid_worker_exit, nullptr), &tv);
printf("worker thread #%d created\n", i);
return context;
}
int main(int argc, char** argv)
{
usage(argv[0]);
......@@ -227,48 +307,57 @@ int main(int argc, char** argv)
}
}
if (argc > 2) {
thread_count = atoi(argv[2]);
if (thread_count <= 0) {
puts("Invalid thread_count");
return 1;
}
}
if (argc > 3) {
disable_data_print = atoi(argv[3]) == 1;
}
sockaddr_in sin = { 0 };
sin.sin_family = AF_INET;
sin.sin_addr.s_addr = htonl(INADDR_ANY);
sin.sin_port = htons(port);
printf("using libevent %s\n", event_get_version());
auto base = event_base_new();
if (!base) {
fprintf(stderr, "init libevent failed\n");
return -1;
printf("%s is listening on port %d using %d threads, disable_data_print=%d\n",
argv[0], port, thread_count, disable_data_print);
std::vector<std::thread> worker_threads;
for (int i = 0; i < thread_count; i++) {
auto context = init_worker_thread(i);
worker_thread_contexts.push_back(context);
worker_threads.emplace_back(std::thread([context]() {
event_base_dispatch(context->base);
}));
}
auto listener = evconnlistener_new_bind(base,
accept_cb,
nullptr,
LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE,
-1, // backlog, -1 for default
(sockaddr*)(&sin),
sizeof(sin));
if (!listener) {
fprintf(stderr, "create listener failed\n");
return -1;
}
printf("%s is listening on port %d\n", argv[0], port);
evconnlistener_set_error_cb(listener, accpet_error_cb);
std::thread t([&base]() {
event_base_dispatch(base);
});
init_listener_thread(sin);
std::thread listener_thread([]() { event_base_dispatch(listen_thread_evbase); });
op_usage();
while (1) {
int cmd = getchar();
auto fire_command = []() {
struct timeval tv = { 0, 1000 };
for (int i = 0; i < thread_count; i++) {
event_add(event_new(worker_thread_contexts[i]->base, -1, 0, commandcb, worker_thread_contexts[i]), &tv);
}
};
while (1) {
int cmd = getchar();
if (cmd == 'a' || cmd == 'A') {
{
std::lock_guard<std::mutex> lg(mutex);
events.push_back(EVENT_ARM);
threads_to_handled_event = thread_count;
}
event_add(event_new(base, -1, 0, commandcb, nullptr), &tv);
fire_command();
} else if (cmd == 'd' || cmd == 'D') {
do {
printf("Input 6 digit password:");
......@@ -277,18 +366,25 @@ int main(int argc, char** argv)
{
std::lock_guard<std::mutex> lg(mutex);
events.push_back(EVENT_DISARM);
threads_to_handled_event = thread_count;
}
event_add(evtimer_new(base, commandcb, nullptr), &tv);
fire_command();
} else if (cmd == 'e' || cmd == 'E') {
{
std::lock_guard<std::mutex> lg(mutex);
events.push_back(EVENT_EMERGENCY);
threads_to_handled_event = thread_count;
}
event_add(evtimer_new(base, commandcb, nullptr), &tv);
fire_command();
} else if (cmd == '\r' || cmd == '\n') {
} else if (cmd == 'q' || cmd == 'Q') {
event_base_loopbreak(base);
t.join();
timeval tv{ 0, 1000 };
event_base_loopexit(listen_thread_evbase, &tv);
listener_thread.join();
for (int i = 0; i < thread_count; i++) {
event_base_loopexit(worker_thread_contexts[i]->base, &tv);
worker_threads[i].join();
}
break;
} else {
op_usage();
......
<?xml version="1.0" encoding="utf-8"?>
<Project ToolsVersion="Current" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
<PropertyGroup />
<PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|x64'">
<LocalDebuggerCommandArguments>12345 8 1</LocalDebuggerCommandArguments>
<DebuggerFlavor>WindowsLocalDebugger</DebuggerFlavor>
</PropertyGroup>
</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