lorenz.stechauner/sesimos
1
1
Fork 0
Code Issues Pull Requests Projects Releases Activity

Fix mpmc worker termination

Browse Source
This commit is contained in:
Lorenz Stechauner
2023-01-03 10:34:08 +01:00
parent 1cf9172194
commit e72cb57b4a
2 changed files with 15 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
src/lib/mpmc.c
View File

@ -8,11 +8,6 @@
#include <pthread.h> #include <pthread.h>
#include <signal.h> #include <signal.h>
typedef struct {
mpmc_t *ctx;
int worker_id;
} mpmc_arg_t;
static void *mpmc_worker(void *arg); static void *mpmc_worker(void *arg);
int mpmc_init(mpmc_t *ctx, int n_workers, int buf_size, void (*consumer)(void *obj), const char *name) { int mpmc_init(mpmc_t *ctx, int n_workers, int buf_size, void (*consumer)(void *obj), const char *name) {
@ -40,8 +35,8 @@ int mpmc_init(mpmc_t *ctx, int n_workers, int buf_size, void (*consumer)(void *o
return -1; return -1;
} }
memset(ctx->buffer, 0, ctx->size * sizeof(void *)); memset(ctx->buffer, 0, ctx->size * sizeof(void *));
memset(ctx->workers, 0, ctx->n_workers * sizeof(pthread_t)); memset(ctx->workers, -1, ctx->n_workers * sizeof(pthread_t));
for (int i = 0; i < ctx->n_workers; i++) { for (int i = 0; i < ctx->n_workers; i++) {
int ret; int ret;
@ -155,11 +150,9 @@ void mpmc_destroy(mpmc_t *ctx) {
// stop threads, if running // stop threads, if running
mpmc_stop(ctx); mpmc_stop(ctx);
for (int i = 0; i < ctx->n_workers; i++) { for (int i = 0; i < ctx->n_workers; i++) {
if (ctx->workers[i] == 0) break; if (ctx->workers[i] == -1) break;
// FIXME
pthread_kill(ctx->workers[i], SIGUSR1); pthread_kill(ctx->workers[i], SIGUSR1);
//pthread_join(ctx->workers[i], NULL); pthread_join(ctx->workers[i], NULL);
pthread_cancel(ctx->workers[i]);
} }
sem_destroy(&ctx->free); sem_destroy(&ctx->free);

17
src/server.c
View File

@ -125,8 +125,10 @@ static void terminate_gracefully(int sig) {
notice("Terminating gracefully..."); notice("Terminating gracefully...");
server_alive = 0; server_alive = 0;
signal(SIGINT, terminate_forcefully); struct sigaction act = {0};
signal(SIGTERM, terminate_forcefully); act.sa_handler = terminate_forcefully;
sigaction(SIGINT, &act, NULL);
sigaction(SIGTERM, &act, NULL);
workers_stop(); workers_stop();
workers_destroy(); workers_destroy();
@ -225,10 +227,13 @@ int main(int argc, char *const argv[]) {
return 1; return 1;
} }
signal(SIGINT, terminate_gracefully); struct sigaction act = {0};
signal(SIGTERM, terminate_gracefully); act.sa_handler = terminate_gracefully;
signal(SIGUSR1, nothing); sigaction(SIGINT, &act, NULL);
signal(SIGPIPE, nothing); sigaction(SIGTERM, &act, NULL);
act.sa_handler = nothing;
sigaction(SIGUSR1, &act, NULL);
sigaction(SIGPIPE, &act, NULL);
if ((ret = geoip_init(config.geoip_dir)) != 0) { if ((ret = geoip_init(config.geoip_dir)) != 0) {
if (ret == -1) { if (ret == -1) {