From 156897c8a6c7b86cb244d74911528de5c28f2134 Mon Sep 17 00:00:00 2001 From: "Steinar H. Gunderson" Date: Wed, 10 Apr 2013 23:45:20 +0200 Subject: [PATCH] Wrap the acceptor into the same thread logic as everything else. --- acceptor.cpp | 33 ++++++++++++++++++++++++++++++--- acceptor.h | 18 +++++++++++++++++- main.cpp | 10 +++------- 3 files changed, 50 insertions(+), 11 deletions(-) diff --git a/acceptor.cpp b/acceptor.cpp index 0b8de50..776f39b 100644 --- a/acceptor.cpp +++ b/acceptor.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include #include #include @@ -59,10 +60,37 @@ int create_server_socket(int port) return server_sock; } + +AcceptorThread::AcceptorThread(int server_sock) + : server_sock(server_sock) +{ +} + +void AcceptorThread::run() +{ + should_stop = false; + pthread_create(&worker_thread, NULL, &AcceptorThread::do_work_thunk, this); +} -void *acceptor_thread_run(void *arg) +void AcceptorThread::stop() +{ + should_stop = true; + pthread_kill(worker_thread, SIGHUP); + if (pthread_join(worker_thread, NULL) == -1) { + perror("pthread_join"); + exit(1); + } +} + +void *AcceptorThread::do_work_thunk(void *arg) +{ + AcceptorThread *acceptor_thread = reinterpret_cast(arg); + acceptor_thread->do_work(); + return NULL; +} + +void AcceptorThread::do_work() { - int server_sock = int(intptr_t(arg)); while (!hupped) { // Since we are non-blocking, we need to wait for the right state first. // Wait up to 50 ms, then check hupped. @@ -104,5 +132,4 @@ void *acceptor_thread_run(void *arg) // Pick a server, round-robin, and hand over the socket to it. servers->add_client(sock); } - return NULL; } diff --git a/acceptor.h b/acceptor.h index f173e5b..5cd579d 100644 --- a/acceptor.h +++ b/acceptor.h @@ -5,6 +5,22 @@ int create_server_socket(int port); // A thread that accepts new connections on a given socket, // and hands them off to the server pool. -void *acceptor_thread_run(void *arg); +class AcceptorThread { +public: + AcceptorThread(int server_sock); + void run(); + void stop(); + +private: + // Recovers the this pointer, and hands over control to do_work(). + static void *do_work_thunk(void *arg); + + void do_work(); + + int server_sock; + + pthread_t worker_thread; + volatile bool should_stop; +}; #endif // !defined(_ACCEPTOR_H) diff --git a/main.cpp b/main.cpp index e270141..79d9839 100644 --- a/main.cpp +++ b/main.cpp @@ -265,8 +265,8 @@ int main(int argc, char **argv) servers->run(); - pthread_t acceptor_thread; - pthread_create(&acceptor_thread, NULL, acceptor_thread_run, reinterpret_cast(server_sock)); + AcceptorThread acceptor_thread(server_sock); + acceptor_thread.run(); // Find all streams in the configuration file, and create inputs for them. vector inputs; @@ -342,11 +342,7 @@ int main(int argc, char **argv) if (stats_thread != NULL) { stats_thread->stop(); } - pthread_kill(acceptor_thread, SIGHUP); - if (pthread_join(acceptor_thread, NULL) == -1) { - perror("pthread_join"); - exit(1); - } + acceptor_thread.stop(); CubemapStateProto state; state.set_serialize_start_sec(serialize_start.tv_sec); -- 2.39.2