/* * Copyright 2024 Vsevolod Stakhov * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ #include "config.h" #include "blas-config.h" #include "rspamd.h" #include "libserver/maps/map.h" #include "lua/lua_common.h" #include "libserver/worker_util.h" #include "libserver/rspamd_control.h" #include "ottery.h" #include "cryptobox.h" #include "utlist.h" #include "unix-std.h" /* pwd and grp */ #ifdef HAVE_PWD_H #include #endif #ifdef HAVE_GRP_H #include #endif #ifdef HAVE_NFTW #include #endif #include #ifdef HAVE_SYS_RESOURCE_H #include #endif #ifdef HAVE_LIBUTIL_H #include #endif #ifdef HAVE_OPENSSL #include #include #include #endif #include "sqlite3.h" #include "contrib/libev/ev.h" #ifdef WITH_HYPERSCAN #include "libserver/hyperscan_tools.h" #endif /* 2 seconds to fork new process in place of dead one */ #define SOFT_FORK_TIME 2 /* 10 seconds after getting termination signal to terminate all workers with SIGKILL */ #define TERMINATION_INTERVAL (0.2) static gboolean load_rspamd_config(struct rspamd_main *rspamd_main, struct rspamd_config *cfg, gboolean init_modules, enum rspamd_post_load_options opts, gboolean reload); static void rspamd_cld_handler(EV_P_ ev_child *w, struct rspamd_main *rspamd_main, struct rspamd_worker *wrk); /* Control socket */ static int control_fd; static ev_io control_ev; static struct rspamd_stat old_stat; static ev_timer stat_ev; static gboolean valgrind_mode = FALSE; /* Cmdline options */ static gboolean no_fork = FALSE; static gboolean show_version = FALSE; static char **cfg_names = NULL; static char *rspamd_user = NULL; static char *rspamd_group = NULL; static char *rspamd_pidfile = NULL; static gboolean is_debug = FALSE; static gboolean is_insecure = FALSE; static GHashTable *ucl_vars = NULL; static char **lua_env = NULL; static gboolean skip_template = FALSE; static int term_attempts = 0; /* List of active listen sockets indexed by worker type */ static GHashTable *listen_sockets = NULL; /* Defined in modules.c */ extern module_t *modules[]; extern worker_t *workers[]; /* Command line options */ static gboolean rspamd_parse_var(const char *option_name, const char *value, gpointer data, GError **error); static GOptionEntry entries[] = { {"no-fork", 'f', 0, G_OPTION_ARG_NONE, &no_fork, "Do not daemonize main process", NULL}, {"config", 'c', 0, G_OPTION_ARG_FILENAME_ARRAY, &cfg_names, "Specify config file(s)", NULL}, {"user", 'u', 0, G_OPTION_ARG_STRING, &rspamd_user, "User to run rspamd as", NULL}, {"group", 'g', 0, G_OPTION_ARG_STRING, &rspamd_group, "Group to run rspamd as", NULL}, {"pid", 'p', 0, G_OPTION_ARG_STRING, &rspamd_pidfile, "Path to pidfile", NULL}, {"debug", 'd', 0, G_OPTION_ARG_NONE, &is_debug, "Force debug output", NULL}, {"insecure", 'i', 0, G_OPTION_ARG_NONE, &is_insecure, "Ignore running workers as privileged users (insecure)", NULL}, {"version", 'v', 0, G_OPTION_ARG_NONE, &show_version, "Show version and exit", NULL}, {"var", 0, 0, G_OPTION_ARG_CALLBACK, (gpointer) &rspamd_parse_var, "Redefine/define environment variable", NULL}, {"skip-template", 'T', 0, G_OPTION_ARG_NONE, &skip_template, "Do not apply Jinja templates", NULL}, {"lua-env", '\0', 0, G_OPTION_ARG_FILENAME_ARRAY, &lua_env, "Load lua environment from the specified files", NULL}, {NULL, 0, 0, G_OPTION_ARG_NONE, NULL, NULL, NULL}}; static gboolean rspamd_parse_var(const char *option_name, const char *value, gpointer data, GError **error) { char *k, *v, *t; t = strchr(value, '='); if (t != NULL) { k = g_strdup(value); t = k + (t - value); v = g_strdup(t + 1); *t = '\0'; if (ucl_vars == NULL) { ucl_vars = g_hash_table_new_full(rspamd_strcase_hash, rspamd_strcase_equal, g_free, g_free); } g_hash_table_insert(ucl_vars, k, v); } else { g_set_error(error, g_quark_try_string("main"), EINVAL, "Bad variable format: %s", value); return FALSE; } return TRUE; } static void read_cmd_line(int *argc, char ***argv, struct rspamd_config *cfg) { GError *error = NULL; GOptionContext *context; unsigned int cfg_num; context = g_option_context_new("- run rspamd daemon"); #if defined(GIT_VERSION) && GIT_VERSION == 1 g_option_context_set_summary(context, "Summary:\n Rspamd daemon version " RVERSION "-git\n Git id: " RID); #else g_option_context_set_summary(context, "Summary:\n Rspamd daemon version " RVERSION); #endif g_option_context_add_main_entries(context, entries, NULL); if (!g_option_context_parse(context, argc, argv, &error)) { fprintf(stderr, "option parsing failed: %s\n", error->message); g_option_context_free(context); exit(EXIT_FAILURE); } cfg->rspamd_user = rspamd_user; cfg->rspamd_group = rspamd_group; cfg_num = cfg_names != NULL ? g_strv_length(cfg_names) : 0; if (cfg_num == 0) { cfg->cfg_name = FIXED_CONFIG_FILE; } else { cfg->cfg_name = cfg_names[0]; g_assert(cfg_num == 1); } cfg->pid_file = rspamd_pidfile; g_option_context_free(context); } static int rspamd_write_pid(struct rspamd_main *main) { pid_t pid; if (main->cfg->pid_file == NULL) { return -1; } main->pfh = rspamd_pidfile_open(main->cfg->pid_file, 0644, &pid); if (main->pfh == NULL) { return -1; } if (main->is_privileged) { /* Force root user as owner of pid file */ #ifdef HAVE_PIDFILE_FILENO if (fchown(pidfile_fileno(main->pfh), 0, 0) == -1) { #else if (fchown(main->pfh->pf_fd, 0, 0) == -1) { #endif } } rspamd_pidfile_write(main->pfh); return 0; } /* Detect privileged mode */ static void detect_priv(struct rspamd_main *rspamd_main) { struct passwd *pwd; struct group *grp; uid_t euid; euid = geteuid(); if (euid == 0) { if (!rspamd_main->cfg->rspamd_user && !is_insecure) { msg_err_main( "cannot run rspamd workers as root user, please add -u and -g options to select a proper unprivileged user or specify --insecure flag"); exit(EXIT_FAILURE); } else if (is_insecure) { rspamd_main->is_privileged = TRUE; rspamd_main->workers_uid = 0; rspamd_main->workers_gid = 0; } else { rspamd_main->is_privileged = TRUE; pwd = getpwnam(rspamd_main->cfg->rspamd_user); if (pwd == NULL) { msg_err_main("user specified does not exists (%s), aborting", strerror(errno)); exit(-errno); } if (rspamd_main->cfg->rspamd_group) { grp = getgrnam(rspamd_main->cfg->rspamd_group); if (grp == NULL) { msg_err_main("group specified does not exists (%s), aborting", strerror(errno)); exit(-errno); } rspamd_main->workers_gid = grp->gr_gid; } else { /* Use the main group of user */ rspamd_main->workers_gid = pwd->pw_gid; } rspamd_main->workers_uid = pwd->pw_uid; } } else { rspamd_main->is_privileged = FALSE; rspamd_main->workers_uid = (uid_t) -1; rspamd_main->workers_gid = (gid_t) -1; } } static void config_logger(rspamd_mempool_t *pool, gpointer ud) { struct rspamd_main *rspamd_main = ud; rspamd_main->logger = rspamd_log_open_specific(rspamd_main->server_pool, rspamd_main->cfg, "main", rspamd_main->workers_uid, rspamd_main->workers_gid); if (rspamd_main->logger == NULL) { /* * XXX: * Error has been already logged (in fact, * we might fall back to console logger here) */ exit(EXIT_FAILURE); } rspamd_logger_configure_modules(rspamd_main->cfg->debug_modules); } static gboolean reread_config(struct rspamd_main *rspamd_main) { struct rspamd_config *tmp_cfg, *old_cfg; char *cfg_file; int load_opts = RSPAMD_CONFIG_INIT_VALIDATE | RSPAMD_CONFIG_INIT_SYMCACHE | RSPAMD_CONFIG_INIT_LIBS | RSPAMD_CONFIG_INIT_URL; rspamd_symcache_save(rspamd_main->cfg->cache); tmp_cfg = rspamd_config_new(RSPAMD_CONFIG_INIT_DEFAULT); tmp_cfg->libs_ctx = rspamd_main->cfg->libs_ctx; REF_RETAIN(tmp_cfg->libs_ctx); cfg_file = rspamd_mempool_strdup(tmp_cfg->cfg_pool, rspamd_main->cfg->cfg_name); /* Save some variables */ tmp_cfg->cfg_name = cfg_file; old_cfg = rspamd_main->cfg; rspamd_main->cfg = tmp_cfg; rspamd_logger_t *old_logger = rspamd_main->logger; if (!load_rspamd_config(rspamd_main, tmp_cfg, TRUE, load_opts, TRUE)) { rspamd_main->cfg = old_cfg; rspamd_main->logger = old_logger; msg_err_main("cannot parse new config file, revert to old one"); REF_RELEASE(tmp_cfg); return FALSE; } else { rspamd_log_close(old_logger); msg_info_main("replacing config"); REF_RELEASE(old_cfg); rspamd_main->cfg->rspamd_user = rspamd_user; rspamd_main->cfg->rspamd_group = rspamd_group; /* Here, we can do post actions with the existing config */ /* * As some rules are defined in lua, we need to process them, then init * modules and merely afterwards to init modules */ rspamd_lua_post_load_config(tmp_cfg); rspamd_init_filters(tmp_cfg, true, false); /* Do post-load actions */ rspamd_config_post_load(tmp_cfg, load_opts | RSPAMD_CONFIG_INIT_POST_LOAD_LUA | RSPAMD_CONFIG_INIT_PRELOAD_MAPS); msg_info_main("config has been reread successfully"); } return TRUE; } struct waiting_worker { struct rspamd_main *rspamd_main; struct ev_timer wait_ev; struct rspamd_worker_conf *cf; unsigned int oldindex; }; static void rspamd_fork_delayed_cb(EV_P_ ev_timer *w, int revents) { struct waiting_worker *waiting_worker = (struct waiting_worker *) w->data; ev_timer_stop(EV_A_ & waiting_worker->wait_ev); rspamd_fork_worker(waiting_worker->rspamd_main, waiting_worker->cf, waiting_worker->oldindex, waiting_worker->rspamd_main->event_loop, rspamd_cld_handler, listen_sockets); REF_RELEASE(waiting_worker->cf); g_free(waiting_worker); } static void rspamd_fork_delayed(struct rspamd_worker_conf *cf, unsigned int index, struct rspamd_main *rspamd_main) { struct waiting_worker *nw; nw = g_malloc0(sizeof(*nw)); nw->cf = cf; nw->oldindex = index; nw->rspamd_main = rspamd_main; REF_RETAIN(cf); nw->wait_ev.data = nw; ev_timer_init(&nw->wait_ev, rspamd_fork_delayed_cb, SOFT_FORK_TIME, 0.0); ev_timer_start(rspamd_main->event_loop, &nw->wait_ev); } static GList * create_listen_socket(GPtrArray *addrs, unsigned int cnt, enum rspamd_worker_socket_type listen_type) { GList *result = NULL; int fd; unsigned int i; static const int listen_opts = RSPAMD_INET_ADDRESS_LISTEN_ASYNC; struct rspamd_worker_listen_socket *ls; g_ptr_array_sort(addrs, rspamd_inet_address_compare_ptr); for (i = 0; i < cnt; i++) { /* * Copy address to avoid reload issues */ if (listen_type & RSPAMD_WORKER_SOCKET_TCP) { fd = rspamd_inet_address_listen(g_ptr_array_index(addrs, i), SOCK_STREAM, listen_opts, -1); if (fd != -1) { ls = g_malloc0(sizeof(*ls)); ls->addr = rspamd_inet_address_copy(g_ptr_array_index(addrs, i), NULL); ls->fd = fd; ls->type = RSPAMD_WORKER_SOCKET_TCP; result = g_list_prepend(result, ls); } } if (listen_type & RSPAMD_WORKER_SOCKET_UDP) { fd = rspamd_inet_address_listen(g_ptr_array_index(addrs, i), SOCK_DGRAM, listen_opts | RSPAMD_INET_ADDRESS_LISTEN_REUSEPORT, -1); if (fd != -1) { ls = g_malloc0(sizeof(*ls)); ls->addr = rspamd_inet_address_copy(g_ptr_array_index(addrs, i), NULL); ls->fd = fd; ls->type = RSPAMD_WORKER_SOCKET_UDP; result = g_list_prepend(result, ls); } } } return result; } static GList * systemd_get_socket(struct rspamd_main *rspamd_main, const char *fdname) { int number, sock, num_passed, flags; GList *result = NULL; const char *e; char **fdnames; char *end; struct stat st; static const int sd_listen_fds_start = 3; /* SD_LISTEN_FDS_START */ struct rspamd_worker_listen_socket *ls; union { struct sockaddr_storage ss; struct sockaddr sa; struct sockaddr_un sun; struct sockaddr_in6 s6; } addr_storage; socklen_t slen = sizeof(addr_storage); int stype; number = strtoul(fdname, &end, 10); if (end != NULL && *end != '\0') { /* Cannot parse as number, assume a name in LISTEN_FDNAMES. */ e = getenv("LISTEN_FDNAMES"); if (!e) { msg_err_main("cannot get systemd variable 'LISTEN_FDNAMES'"); errno = ENOENT; return NULL; } fdnames = g_strsplit(e, ":", -1); for (number = 0; fdnames[number]; number++) { if (!strcmp(fdnames[number], fdname)) { break; } } if (!fdnames[number]) { number = -1; } g_strfreev(fdnames); } if (number < 0) { msg_warn_main("cannot find systemd socket: %s", fdname); errno = ENOENT; return NULL; } e = getenv("LISTEN_FDS"); if (e != NULL) { errno = 0; num_passed = strtoul(e, &end, 10); if ((end == NULL || *end == '\0') && num_passed > number) { sock = number + sd_listen_fds_start; if (fstat(sock, &st) == -1) { msg_warn_main("cannot stat systemd descriptor %d", sock); return NULL; } if (!S_ISSOCK(st.st_mode)) { msg_warn_main("systemd descriptor %d is not a socket", sock); errno = EINVAL; return NULL; } flags = fcntl(sock, F_GETFD); if (flags != -1) { (void) fcntl(sock, F_SETFD, flags | FD_CLOEXEC); } rspamd_socket_nonblocking(sock); if (getsockname(sock, &addr_storage.sa, &slen) == -1) { msg_warn_main("cannot get name for systemd descriptor %d: %s", sock, strerror(errno)); errno = EINVAL; return NULL; } ls = g_malloc0(sizeof(*ls)); ls->addr = rspamd_inet_address_from_sa(&addr_storage.sa, slen); ls->fd = sock; ls->is_systemd = true; slen = sizeof(stype); if (getsockopt(sock, SOL_SOCKET, SO_TYPE, &stype, &slen) != -1) { if (stype == SOCK_STREAM) { ls->type = RSPAMD_WORKER_SOCKET_TCP; } else { ls->type = RSPAMD_WORKER_SOCKET_UDP; } } else { msg_warn_main("cannot get type for systemd descriptor %d: %s", sock, strerror(errno)); ls->type = RSPAMD_WORKER_SOCKET_TCP; } result = g_list_prepend(result, ls); } else if (num_passed <= number) { msg_err_main("systemd LISTEN_FDS does not contain the expected fd: %d", num_passed); errno = EINVAL; } } else { msg_err_main("cannot get systemd variable 'LISTEN_FDS'"); errno = ENOENT; } return result; } static void pass_signal_cb(gpointer key, gpointer value, gpointer ud) { struct rspamd_worker *cur = value; int signo = GPOINTER_TO_INT(ud); kill(cur->pid, signo); } static void rspamd_pass_signal(GHashTable *workers, int signo) { g_hash_table_foreach(workers, pass_signal_cb, GINT_TO_POINTER(signo)); } static inline uintptr_t make_listen_key(struct rspamd_worker_bind_conf *cf) { rspamd_cryptobox_fast_hash_state_t st; unsigned int i, keylen = 0; uint8_t *key; rspamd_inet_addr_t *addr; uint16_t port; rspamd_cryptobox_fast_hash_init(&st, rspamd_hash_seed()); if (cf->is_systemd) { /* Something like 'systemd:0' or 'systemd:controller'. */ rspamd_cryptobox_fast_hash_update(&st, cf->name, strlen(cf->name)); } else { rspamd_cryptobox_fast_hash_update(&st, cf->name, strlen(cf->name)); for (i = 0; i < cf->cnt; i++) { addr = g_ptr_array_index(cf->addrs, i); key = rspamd_inet_address_get_hash_key( addr, &keylen); rspamd_cryptobox_fast_hash_update(&st, key, keylen); port = rspamd_inet_address_get_port(addr); rspamd_cryptobox_fast_hash_update(&st, &port, sizeof(port)); } } return rspamd_cryptobox_fast_hash_final(&st); } static void spawn_worker_type(struct rspamd_main *rspamd_main, struct ev_loop *event_loop, struct rspamd_worker_conf *cf) { int i; if (cf->count < 0) { msg_info_main("skip spawning of worker %s: disabled in configuration", cf->worker->name); return; } if (cf->worker->flags & RSPAMD_WORKER_UNIQUE) { if (cf->count > 1) { msg_warn_main( "cannot spawn more than 1 %s worker, so spawn one", cf->worker->name); } rspamd_fork_worker(rspamd_main, cf, 0, event_loop, rspamd_cld_handler, listen_sockets); } else if (cf->worker->flags & RSPAMD_WORKER_THREADED) { rspamd_fork_worker(rspamd_main, cf, 0, event_loop, rspamd_cld_handler, listen_sockets); } else { for (i = 0; i < cf->count; i++) { rspamd_fork_worker(rspamd_main, cf, i, event_loop, rspamd_cld_handler, listen_sockets); } } } static void spawn_workers(struct rspamd_main *rspamd_main, struct ev_loop *ev_base) { GList *cur, *ls; struct rspamd_worker_conf *cf; gpointer p; guintptr key; struct rspamd_worker_bind_conf *bcf; gboolean listen_ok = FALSE; GPtrArray *seen_mandatory_workers; worker_t **cw, *wrk; unsigned int i; /* Special hack for hs_helper if it's not defined in a config */ seen_mandatory_workers = g_ptr_array_new(); cur = rspamd_main->cfg->workers; while (cur) { cf = cur->data; listen_ok = FALSE; if (cf->worker == NULL) { msg_err_main("type of worker is unspecified, skip spawning"); } else { if (!cf->enabled || cf->count <= 0) { msg_info_main("worker of type %s(%s) is disabled in the config, " "skip spawning", g_quark_to_string(cf->type), cf->bind_conf ? cf->bind_conf->name : "none"); cur = g_list_next(cur); continue; } if (cf->worker->flags & RSPAMD_WORKER_ALWAYS_START) { g_ptr_array_add(seen_mandatory_workers, cf->worker); } if (cf->worker->flags & RSPAMD_WORKER_HAS_SOCKET) { LL_FOREACH(cf->bind_conf, bcf) { key = make_listen_key(bcf); if ((p = g_hash_table_lookup(listen_sockets, GINT_TO_POINTER(key))) == NULL) { if (!bcf->is_systemd) { /* Create listen socket */ ls = create_listen_socket(bcf->addrs, bcf->cnt, cf->worker->listen_type); } else { ls = systemd_get_socket(rspamd_main, g_ptr_array_index(bcf->addrs, 0)); } if (ls == NULL) { msg_err_main("cannot listen on %s socket %s: %s", bcf->is_systemd ? "systemd" : "normal", bcf->name, strerror(errno)); } else { g_hash_table_insert(listen_sockets, (gpointer) key, ls); listen_ok = TRUE; } } else { /* We had socket for this type of worker */ ls = p; listen_ok = TRUE; } /* Do not add existing lists as it causes loops */ if (g_list_position(cf->listen_socks, ls) == -1) { cf->listen_socks = g_list_concat(cf->listen_socks, ls); } } if (listen_ok) { spawn_worker_type(rspamd_main, ev_base, cf); } else { if (cf->bind_conf == NULL) { msg_err_main("cannot create listen socket for %s", g_quark_to_string(cf->type)); } else { msg_err_main("cannot create listen socket for %s at %s", g_quark_to_string(cf->type), cf->bind_conf->name); } rspamd_hard_terminate(rspamd_main); g_assert_not_reached(); } } else { spawn_worker_type(rspamd_main, ev_base, cf); } } cur = g_list_next(cur); } for (cw = workers; *cw != NULL; cw++) { gboolean seen = FALSE; wrk = *cw; if (wrk->flags & RSPAMD_WORKER_ALWAYS_START) { for (i = 0; i < seen_mandatory_workers->len; i++) { if (wrk == g_ptr_array_index(seen_mandatory_workers, i)) { seen = TRUE; break; } } if (!seen) { cf = rspamd_config_new_worker(rspamd_main->cfg, NULL); cf->count = 1; cf->worker = wrk; cf->type = g_quark_from_static_string(wrk->name); if (cf->worker->worker_init_func) { cf->ctx = cf->worker->worker_init_func(rspamd_main->cfg); } spawn_worker_type(rspamd_main, ev_base, cf); } } } g_ptr_array_free(seen_mandatory_workers, TRUE); } static void kill_old_workers(gpointer key, gpointer value, gpointer unused) { struct rspamd_worker *w = value; struct rspamd_main *rspamd_main; rspamd_main = w->srv; if (w->state == rspamd_worker_state_wanna_die) { w->state = rspamd_worker_state_terminating; kill(w->pid, SIGUSR2); ev_io_stop(rspamd_main->event_loop, &w->srv_ev); g_hash_table_remove_all(w->control_events_pending); msg_info_main("send signal to worker %P", w->pid); } else if (w->state != rspamd_worker_state_running) { msg_info_main("do not send signal to worker %P, already sent", w->pid); } } static void mark_old_workers(gpointer key, gpointer value, gpointer unused) { struct rspamd_worker *w = value; if (w->state == rspamd_worker_state_running) { w->state = rspamd_worker_state_wanna_die; } w->flags |= RSPAMD_WORKER_OLD_CONFIG; } static void rspamd_worker_wait(struct rspamd_worker *w) { struct rspamd_main *rspamd_main; rspamd_main = w->srv; if (term_attempts < 0) { if (w->cf->worker->flags & RSPAMD_WORKER_KILLABLE) { if (kill(w->pid, SIGKILL) == -1) { if (errno == ESRCH) { /* We have actually killed the process */ return; } } else { msg_warn_main("terminate worker %s(%P) with SIGKILL", g_quark_to_string(w->type), w->pid); } } else { kill(w->pid, SIGKILL); if (errno == ESRCH) { /* We have actually killed the process */ return; } else { msg_err_main("data corruption warning: terminating " "special worker %s(%P) with SIGKILL", g_quark_to_string(w->type), w->pid); } } } } static void hash_worker_wait_callback(gpointer key, gpointer value, gpointer unused) { rspamd_worker_wait((struct rspamd_worker *) value); } struct core_check_cbdata { struct rspamd_config *cfg; gsize total_count; gsize total_size; }; #ifdef HAVE_NFTW static struct core_check_cbdata cores_cbdata; static int rspamd_check_core_cb(const char *path, const struct stat *st, int flag, struct FTW *ft) { if (S_ISREG(st->st_mode)) { cores_cbdata.total_count++; /* Use physical size instead of displayed one */ cores_cbdata.total_size += st->st_blocks * 512; } return 0; } #endif static void rspamd_check_core_limits(struct rspamd_main *rspamd_main) { #ifdef HAVE_NFTW struct rspamd_config *cfg = rspamd_main->cfg; cores_cbdata.cfg = cfg; cores_cbdata.total_count = 0; cores_cbdata.total_size = 0; if (cfg->cores_dir && (cfg->max_cores_count || cfg->max_cores_size)) { if (nftw(cfg->cores_dir, rspamd_check_core_cb, 1, FTW_MOUNT | FTW_PHYS) == -1) { msg_err_main("nftw failed for path %s: %s", cfg->cores_dir, strerror(errno)); } else { if (!rspamd_main->cores_throttling) { if (cfg->max_cores_size && cores_cbdata.total_size > cfg->max_cores_size) { msg_warn_main( "enable cores throttling as size of cores in" " %s is %Hz, limit is %Hz", cfg->cores_dir, cores_cbdata.total_size, cfg->max_cores_size); rspamd_main->cores_throttling = TRUE; } if (cfg->max_cores_count && cores_cbdata.total_count > cfg->max_cores_count) { msg_warn_main( "enable cores throttling as count of cores in" " %s is %z, limit is %z", cfg->cores_dir, cores_cbdata.total_count, cfg->max_cores_count); rspamd_main->cores_throttling = TRUE; } } else { if (cfg->max_cores_size && cores_cbdata.total_size < cfg->max_cores_size) { msg_info_main( "disable cores throttling as size of cores in" " %s is now %Hz, limit is %Hz", cfg->cores_dir, cores_cbdata.total_size, cfg->max_cores_size); rspamd_main->cores_throttling = FALSE; } if (cfg->max_cores_count && cores_cbdata.total_count < cfg->max_cores_count) { msg_info_main( "disable cores throttling as count of cores in" " %s is %z, limit is %z", cfg->cores_dir, cores_cbdata.total_count, cfg->max_cores_count); rspamd_main->cores_throttling = FALSE; } } } } #endif } static void reopen_log_handler(gpointer key, gpointer value, gpointer unused) { struct rspamd_worker *w = value; struct rspamd_main *rspamd_main; rspamd_main = w->srv; if (kill(w->pid, SIGUSR1) == -1) { msg_err_main("kill failed for pid %P: %s", w->pid, strerror(errno)); } } static gboolean load_rspamd_config(struct rspamd_main *rspamd_main, struct rspamd_config *cfg, gboolean init_modules, enum rspamd_post_load_options opts, gboolean reload) { cfg->compiled_modules = modules; cfg->compiled_workers = workers; if (!rspamd_config_read(cfg, cfg->cfg_name, config_logger, rspamd_main, ucl_vars, skip_template, lua_env)) { return FALSE; } /* Strictly set temp dir */ if (!cfg->temp_dir) { msg_warn_main("tempdir is not set, trying to use $TMPDIR"); cfg->temp_dir = rspamd_mempool_strdup(cfg->cfg_pool, getenv("TMPDIR")); if (!cfg->temp_dir) { msg_warn_main("$TMPDIR is empty too, using /tmp as default"); cfg->temp_dir = rspamd_mempool_strdup(cfg->cfg_pool, "/tmp"); } } if (!reload) { /* * As some rules are defined in lua, we need to process them, then init * modules and merely afterwards to init modules */ rspamd_lua_post_load_config(cfg); if (init_modules) { if (!rspamd_init_filters(cfg, reload, false)) { return FALSE; } } /* Do post-load actions */ if (!rspamd_config_post_load(cfg, opts)) { return FALSE; } } return TRUE; } static void rspamd_detach_worker(struct rspamd_main *rspamd_main, struct rspamd_worker *wrk) { ev_io_stop(rspamd_main->event_loop, &wrk->srv_ev); ev_timer_stop(rspamd_main->event_loop, &wrk->hb.heartbeat_ev); } static void rspamd_attach_worker(struct rspamd_main *rspamd_main, struct rspamd_worker *wrk) { ev_io_start(rspamd_main->event_loop, &wrk->srv_ev); ev_timer_start(rspamd_main->event_loop, &wrk->hb.heartbeat_ev); } static void stop_srv_ev(gpointer key, gpointer value, gpointer ud) { struct rspamd_worker *cur = (struct rspamd_worker *) value; struct rspamd_main *rspamd_main = (struct rspamd_main *) ud; rspamd_detach_worker(rspamd_main, cur); } static void start_srv_ev(gpointer key, gpointer value, gpointer ud) { struct rspamd_worker *cur = (struct rspamd_worker *) value; struct rspamd_main *rspamd_main = (struct rspamd_main *) ud; rspamd_attach_worker(rspamd_main, cur); } static void rspamd_final_timer_handler(EV_P_ ev_timer *w, int revents) { struct rspamd_main *rspamd_main = (struct rspamd_main *) w->data; term_attempts--; g_hash_table_foreach(rspamd_main->workers, hash_worker_wait_callback, NULL); if (g_hash_table_size(rspamd_main->workers) == 0) { ev_break(rspamd_main->event_loop, EVBREAK_ALL); } } /* Signal handlers */ static void rspamd_term_handler(struct ev_loop *loop, ev_signal *w, int revents) { struct rspamd_main *rspamd_main = (struct rspamd_main *) w->data; static ev_timer ev_finale; ev_tstamp shutdown_ts; if (!rspamd_main->wanna_die) { rspamd_main->wanna_die = TRUE; shutdown_ts = MAX(SOFT_SHUTDOWN_TIME, rspamd_main->cfg->task_timeout * 2.0); msg_info_main("catch termination signal, waiting for %d children for %.2f seconds", (int) g_hash_table_size(rspamd_main->workers), valgrind_mode ? shutdown_ts * 10 : shutdown_ts); /* Stop srv events to avoid false notifications */ g_hash_table_foreach(rspamd_main->workers, stop_srv_ev, rspamd_main); rspamd_pass_signal(rspamd_main->workers, SIGTERM); if (control_fd != -1) { ev_io_stop(rspamd_main->event_loop, &control_ev); close(control_fd); } if (valgrind_mode) { /* Special case if we are likely running with valgrind */ term_attempts = shutdown_ts / TERMINATION_INTERVAL * 10; } else { term_attempts = shutdown_ts / TERMINATION_INTERVAL; } ev_finale.data = rspamd_main; ev_timer_init(&ev_finale, rspamd_final_timer_handler, TERMINATION_INTERVAL, TERMINATION_INTERVAL); ev_timer_start(rspamd_main->event_loop, &ev_finale); } } static void rspamd_usr1_handler(struct ev_loop *loop, ev_signal *w, int revents) { struct rspamd_main *rspamd_main = (struct rspamd_main *) w->data; if (!rspamd_main->wanna_die) { rspamd_log_reopen(rspamd_main->logger, rspamd_main->cfg, rspamd_main->workers_uid, rspamd_main->workers_gid); msg_info_main("logging reinitialised"); g_hash_table_foreach(rspamd_main->workers, reopen_log_handler, NULL); } } static void rspamd_stat_update_handler(struct ev_loop *loop, ev_timer *w, int revents) { struct rspamd_main *rspamd_main = (struct rspamd_main *) w->data; struct rspamd_stat cur_stat; char proctitle[128]; memcpy(&cur_stat, rspamd_main->stat, sizeof(cur_stat)); if (old_stat.messages_scanned > 0 && cur_stat.messages_scanned > old_stat.messages_scanned) { double rate = (double) (cur_stat.messages_scanned - old_stat.messages_scanned) / w->repeat; double old_spam = old_stat.actions_stat[METRIC_ACTION_REJECT] + old_stat.actions_stat[METRIC_ACTION_ADD_HEADER] + old_stat.actions_stat[METRIC_ACTION_REWRITE_SUBJECT]; double old_ham = old_stat.actions_stat[METRIC_ACTION_NOACTION]; double new_spam = cur_stat.actions_stat[METRIC_ACTION_REJECT] + cur_stat.actions_stat[METRIC_ACTION_ADD_HEADER] + cur_stat.actions_stat[METRIC_ACTION_REWRITE_SUBJECT]; double new_ham = cur_stat.actions_stat[METRIC_ACTION_NOACTION]; gsize cnt = MAX_AVG_TIME_SLOTS; float sum = rspamd_sum_floats(cur_stat.avg_time.avg_time, &cnt); rspamd_snprintf(proctitle, sizeof(proctitle), "main process; %.1f msg/sec, %.1f msg/sec spam, %.1f msg/sec ham; %.2fs avg processing time", rate, (new_spam - old_spam) / w->repeat, (new_ham - old_ham) / w->repeat, cnt > 0 ? sum / cnt : 0); rspamd_setproctitle("%s", proctitle); } memcpy(&old_stat, &cur_stat, sizeof(cur_stat)); } static void rspamd_hup_handler(struct ev_loop *loop, ev_signal *w, int revents) { struct rspamd_main *rspamd_main = (struct rspamd_main *) w->data; if (!rspamd_main->wanna_die) { msg_info_main("rspamd " RVERSION " is requested to reload configuration"); /* Detach existing workers and stop their heartbeats */ g_hash_table_foreach(rspamd_main->workers, stop_srv_ev, rspamd_main); if (reread_config(rspamd_main)) { rspamd_check_core_limits(rspamd_main); /* Mark old workers */ g_hash_table_foreach(rspamd_main->workers, mark_old_workers, NULL); msg_info_main("spawn workers with a new config"); spawn_workers(rspamd_main, rspamd_main->event_loop); msg_info_main("workers spawning has been finished"); /* Kill marked */ msg_info_main("kill old workers"); g_hash_table_foreach(rspamd_main->workers, kill_old_workers, NULL); } else { /* Reattach old workers */ msg_info_main("restore old workers with a old config"); g_hash_table_foreach(rspamd_main->workers, start_srv_ev, rspamd_main); } } } /* Called when a dead child has been found */ static void rspamd_cld_handler(EV_P_ ev_child *w, struct rspamd_main *rspamd_main, struct rspamd_worker *wrk) { gboolean need_refork; static struct rspamd_control_command cmd; /* Turn off locking for logger */ ev_child_stop(EV_A_ w); /* Remove dead child form children list */ g_hash_table_remove(rspamd_main->workers, GSIZE_TO_POINTER(wrk->pid)); g_hash_table_remove_all(wrk->control_events_pending); if (wrk->srv_pipe[0] != -1) { /* Ugly workaround */ if (wrk->tmp_data) { g_free(wrk->tmp_data); } rspamd_detach_worker(rspamd_main, wrk); } if (wrk->control_pipe[0] != -1) { /* We also need to clean descriptors left */ close(wrk->control_pipe[0]); close(wrk->srv_pipe[0]); } if (!rspamd_main->wanna_die) { cmd.type = RSPAMD_CONTROL_CHILD_CHANGE; cmd.cmd.child_change.what = rspamd_child_terminated; cmd.cmd.child_change.pid = wrk->pid; cmd.cmd.child_change.additional = w->rstatus; rspamd_control_broadcast_srv_cmd(rspamd_main, &cmd, wrk->pid); } need_refork = rspamd_check_termination_clause(wrk->srv, wrk, w->rstatus); if (need_refork) { /* Fork another worker in replace of dead one */ msg_info_main("respawn process %s in lieu of terminated process with pid %P", g_quark_to_string(wrk->type), wrk->pid); rspamd_check_core_limits(rspamd_main); rspamd_fork_delayed(wrk->cf, wrk->index, rspamd_main); } else { msg_info_main("do not respawn process %s after found terminated process with pid %P", g_quark_to_string(wrk->type), wrk->pid); } REF_RELEASE(wrk->cf); g_hash_table_unref(wrk->control_events_pending); g_free(wrk); } /* Control socket handler */ static void rspamd_control_handler(EV_P_ ev_io *w, int revents) { struct rspamd_main *rspamd_main = (struct rspamd_main *) w->data; rspamd_inet_addr_t *addr = NULL; int nfd; if ((nfd = rspamd_accept_from_socket(w->fd, &addr, NULL, NULL)) == -1) { msg_warn_main("accept failed: %s", strerror(errno)); return; } /* Check for EAGAIN */ if (nfd == 0) { rspamd_inet_address_free(addr); return; } msg_info_main("accepted control connection from %s", rspamd_inet_address_to_string(addr)); rspamd_control_process_client_socket(rspamd_main, nfd, addr); } static unsigned int rspamd_spair_hash(gconstpointer p) { return rspamd_cryptobox_fast_hash(p, PAIR_ID_LEN, rspamd_hash_seed()); } static gboolean rspamd_spair_equal(gconstpointer a, gconstpointer b) { return memcmp(a, b, PAIR_ID_LEN) == 0; } static void rspamd_spair_close(gpointer p) { int *fds = p; close(fds[0]); close(fds[1]); g_free(p); } const char * get_cpu_architecture(void) { #if defined(__x86_64__) || defined(_M_X64) return "x86_64"; #elif defined(__i386) || defined(_M_IX86) return "x86"; #elif defined(__aarch64__) return "ARM64"; #elif defined(__arm__) || defined(_M_ARM) return "ARM"; #elif defined(__loongarch__) || defined(__loongarch64) return "LOONGARCH64"; #elif defined(__mips__) return "MIPS"; #elif defined(__powerpc__) || defined(_M_PPC) return "PowerPC"; #elif defined(__sparc__) return "SPARC"; #else return "Unknown"; #endif } static void version(struct rspamd_main *rspamd_main) { #if defined(GIT_VERSION) && GIT_VERSION == 1 rspamd_printf("Rspamd daemon version " RVERSION "-git." RID "\n\n"); #else rspamd_printf("Rspamd daemon version " RVERSION "\n\n"); #endif rspamd_printf("CPU architecture %s; features: %s\n", get_cpu_architecture(), rspamd_main->cfg->libs_ctx->crypto_ctx->cpu_extensions); #ifdef WITH_HYPERSCAN rspamd_printf("Hyperscan enabled: TRUE\n"); #else rspamd_printf("Hyperscan enabled: FALSE\n"); #endif #ifdef WITH_JEMALLOC rspamd_printf("Jemalloc enabled: TRUE\n"); #else rspamd_printf("Jemalloc enabled: FALSE\n"); #endif #ifdef WITH_LUAJIT rspamd_printf("LuaJIT enabled: TRUE (LuaJIT version: %s)\n", LUAJIT_VERSION); #else rspamd_printf("LuaJIT enabled: FALSE (Lua version: %s)\n", LUA_VERSION); #endif #ifndef __has_feature #define __has_feature(x) 0 #endif #if (defined(__has_feature) && __has_feature(address_sanitizer)) || defined(ADDRESS_SANITIZER) rspamd_printf("ASAN enabled: TRUE\n"); #else rspamd_printf("ASAN enabled: FALSE\n"); #endif #ifdef HAVE_CBLAS rspamd_printf("BLAS enabled: TRUE\n"); #else rspamd_printf("BLAS enabled: FALSE\n"); #endif #ifdef WITH_FASTTEXT rspamd_printf("Fasttext enabled: TRUE\n"); #else rspamd_printf("Fasttext enabled: FALSE\n"); #endif } static gboolean rspamd_main_daemon(struct rspamd_main *rspamd_main) { int fd; pid_t old_pid = getpid(); switch (fork()) { case -1: msg_err_main("fork() failed: %s", strerror(errno)); return FALSE; case 0: break; default: /* Old process */ exit(0); } rspamd_log_on_fork(g_quark_from_static_string("main"), rspamd_main->cfg, rspamd_main->logger); if (setsid() == -1) { msg_err_main("setsid () failed: %s", strerror(errno)); return FALSE; } umask(0); fd = open("/dev/null", O_RDWR); if (fd == -1) { msg_err_main("open(\"/dev/null\") failed: %s", strerror(errno)); return FALSE; } if (dup2(fd, STDIN_FILENO) == -1) { msg_err_main("dup2(STDIN) failed: %s", strerror(errno)); return FALSE; } if (dup2(fd, STDOUT_FILENO) == -1) { msg_err_main("dup2(STDOUT) failed: %s", strerror(errno)); return FALSE; } if (fd > STDERR_FILENO) { if (close(fd) == -1) { msg_err_main("close() failed: %s", strerror(errno)); return FALSE; } } msg_info_main("daemonized successfully; old pid %P, new pid %P; pid file: %s", old_pid, getpid(), rspamd_main->cfg->pid_file); return TRUE; } int main(int argc, char **argv, char **env) { int i, res = 0; struct sigaction signals, sigpipe_act; worker_t **pworker; GQuark type; rspamd_inet_addr_t *control_addr = NULL; struct ev_loop *event_loop; struct rspamd_main *rspamd_main; gboolean skip_pid = FALSE; sigset_t control_signals; /* Block special signals on loading */ sigemptyset(&control_signals); sigaddset(&control_signals, SIGHUP); sigaddset(&control_signals, SIGUSR1); sigaddset(&control_signals, SIGUSR2); sigprocmask(SIG_BLOCK, &control_signals, NULL); rspamd_main = (struct rspamd_main *) g_malloc0(sizeof(struct rspamd_main)); rspamd_main->server_pool = rspamd_mempool_new(rspamd_mempool_suggest_size(), "main", 0); rspamd_main->stat = rspamd_mempool_alloc0_shared_(rspamd_main->server_pool, sizeof(struct rspamd_stat), RSPAMD_ALIGNOF(struct rspamd_stat), G_STRLOC); /* Set all time slots to nan */ for (i = 0; i < MAX_AVG_TIME_SLOTS; i++) { rspamd_main->stat->avg_time.avg_time[i] = NAN; } rspamd_main->cfg = rspamd_config_new(RSPAMD_CONFIG_INIT_DEFAULT); rspamd_main->spairs = g_hash_table_new_full(rspamd_spair_hash, rspamd_spair_equal, g_free, rspamd_spair_close); rspamd_main->start_mtx = rspamd_mempool_get_mutex(rspamd_main->server_pool); if (getenv("VALGRIND") != NULL) { valgrind_mode = TRUE; } #ifndef HAVE_SETPROCTITLE rspamd_init_title(rspamd_main->server_pool, argc, argv, env); #endif rspamd_main->cfg->libs_ctx = rspamd_init_libs(); memset(&signals, 0, sizeof(struct sigaction)); read_cmd_line(&argc, &argv, rspamd_main->cfg); if (show_version) { version(rspamd_main); exit(EXIT_SUCCESS); } if (argc > 0) { /* Parse variables */ for (i = 0; i < argc; i++) { if (strchr(argv[i], '=') != NULL) { char *k, *v, *t; k = g_strdup(argv[i]); t = strchr(k, '='); v = g_strdup(t + 1); *t = '\0'; if (ucl_vars == NULL) { ucl_vars = g_hash_table_new_full(rspamd_strcase_hash, rspamd_strcase_equal, g_free, g_free); } g_hash_table_insert(ucl_vars, k, v); } } } if (is_debug) { rspamd_main->cfg->log_level = G_LOG_LEVEL_DEBUG; } else { rspamd_main->cfg->log_level = G_LOG_LEVEL_MESSAGE; } type = g_quark_from_static_string("main"); /* First set logger to console logger */ rspamd_main->logger = rspamd_log_open_emergency(rspamd_main->server_pool, 0); g_assert(rspamd_main->logger != NULL); if (is_debug) { rspamd_log_set_log_level(rspamd_main->logger, G_LOG_LEVEL_DEBUG); } else { rspamd_log_set_log_level(rspamd_main->logger, G_LOG_LEVEL_MESSAGE); } g_log_set_default_handler(rspamd_glib_log_function, rspamd_main->logger); g_set_printerr_handler(rspamd_glib_printerr_function); detect_priv(rspamd_main); msg_notice_main("rspamd " RVERSION " is loading configuration, build id: " RID); pworker = &workers[0]; while (*pworker) { /* Init string quarks */ (void) g_quark_from_static_string((*pworker)->name); pworker++; } /* Init listen sockets hash */ listen_sockets = g_hash_table_new(g_direct_hash, g_direct_equal); sqlite3_initialize(); /* Load config */ if (!load_rspamd_config(rspamd_main, rspamd_main->cfg, TRUE, RSPAMD_CONFIG_LOAD_ALL, FALSE)) { exit(EXIT_FAILURE); } /* Override pidfile from configuration by command line argument */ if (rspamd_pidfile != NULL) { rspamd_main->cfg->pid_file = rspamd_pidfile; } /* Force debug log */ if (is_debug) { rspamd_log_set_log_level(rspamd_main->logger, G_LOG_LEVEL_DEBUG); } /* Create rolling history */ rspamd_main->history = rspamd_roll_history_new(rspamd_main->server_pool, rspamd_main->cfg->history_rows, rspamd_main->cfg); msg_info_main("rspamd " RVERSION " is starting, build id: " RID); rspamd_main->cfg->cfg_name = rspamd_mempool_strdup( rspamd_main->cfg->cfg_pool, rspamd_main->cfg->cfg_name); msg_info_main("cpu features: %s", rspamd_main->cfg->libs_ctx->crypto_ctx->cpu_extensions); msg_info_main("cryptobox configuration: curve25519(libsodium), " "chacha20(%s), poly1305(libsodium), siphash(libsodium), blake2(libsodium), base64(%s)", rspamd_main->cfg->libs_ctx->crypto_ctx->chacha20_impl, rspamd_main->cfg->libs_ctx->crypto_ctx->base64_impl); msg_info_main("libottery prf: %s", ottery_get_impl_name()); /* Daemonize */ if (!no_fork) { if (!rspamd_main_daemon(rspamd_main)) { exit(EXIT_FAILURE); } /* Close emergency logger */ rspamd_log_close(rspamd_log_emergency_logger()); } /* Write info */ rspamd_main->pid = getpid(); rspamd_main->type = type; rspamd_set_crash_handler(rspamd_main); /* Ignore SIGPIPE as we handle write errors manually */ sigemptyset(&sigpipe_act.sa_mask); sigaddset(&sigpipe_act.sa_mask, SIGPIPE); sigpipe_act.sa_handler = SIG_IGN; sigpipe_act.sa_flags = 0; sigaction(SIGPIPE, &sigpipe_act, NULL); if (rspamd_main->cfg->pid_file == NULL) { msg_info_main("pid file is not specified, skipping writing it"); skip_pid = TRUE; } else if (no_fork) { msg_info_main("skip writing pid in no-fork mode"); skip_pid = TRUE; } else if (rspamd_write_pid(rspamd_main) == -1) { msg_err_main("cannot write pid file %s", rspamd_main->cfg->pid_file); exit(-errno); } sigprocmask(SIG_BLOCK, &signals.sa_mask, NULL); /* Set title */ rspamd_setproctitle("main process"); /* Open control socket if needed */ control_fd = -1; if (rspamd_main->cfg->control_socket_path) { if (!rspamd_parse_inet_address(&control_addr, rspamd_main->cfg->control_socket_path, strlen(rspamd_main->cfg->control_socket_path), RSPAMD_INET_ADDRESS_PARSE_DEFAULT)) { msg_err_main("cannot parse inet address %s", rspamd_main->cfg->control_socket_path); } else { control_fd = rspamd_inet_address_listen(control_addr, SOCK_STREAM, RSPAMD_INET_ADDRESS_LISTEN_ASYNC, -1); if (control_fd == -1) { msg_err_main("cannot open control socket at path: %s", rspamd_main->cfg->control_socket_path); } } } /* Maybe read roll history */ if (rspamd_main->history && rspamd_main->cfg->history_file) { rspamd_roll_history_load(rspamd_main->history, rspamd_main->cfg->history_file); } /* Init workers hash */ rspamd_main->workers = g_hash_table_new(g_direct_hash, g_direct_equal); /* Unblock control signals */ sigprocmask(SIG_UNBLOCK, &control_signals, NULL); /* Init event base */ event_loop = ev_default_loop(rspamd_config_ev_backend_get(rspamd_main->cfg)); rspamd_main->event_loop = event_loop; if (event_loop) { int loop_type = ev_backend(event_loop); gboolean effective_backend; const char *loop_str; loop_str = rspamd_config_ev_backend_to_string(loop_type, &effective_backend); if (!effective_backend) { msg_warn_main("event loop uses non-optimal backend: %s", loop_str); } else { msg_info_main("event loop initialised with backend: %s", loop_str); } } else { msg_err("cannot init event loop! exiting"); exit(EXIT_FAILURE); } /* Unblock signals */ sigemptyset(&signals.sa_mask); sigprocmask(SIG_SETMASK, &signals.sa_mask, NULL); /* Set events for signals */ ev_signal_init(&rspamd_main->term_ev, rspamd_term_handler, SIGTERM); rspamd_main->term_ev.data = rspamd_main; ev_signal_start(event_loop, &rspamd_main->term_ev); ev_signal_init(&rspamd_main->int_ev, rspamd_term_handler, SIGINT); rspamd_main->int_ev.data = rspamd_main; ev_signal_start(event_loop, &rspamd_main->int_ev); ev_signal_init(&rspamd_main->hup_ev, rspamd_hup_handler, SIGHUP); rspamd_main->hup_ev.data = rspamd_main; ev_signal_start(event_loop, &rspamd_main->hup_ev); ev_signal_init(&rspamd_main->usr1_ev, rspamd_usr1_handler, SIGUSR1); rspamd_main->usr1_ev.data = rspamd_main; ev_signal_start(event_loop, &rspamd_main->usr1_ev); /* Update proctitle according to number of messages processed */ static const ev_tstamp stat_update_time = 10.0; memset(&old_stat, 0, sizeof(old_stat)); stat_ev.data = rspamd_main; ev_timer_init(&stat_ev, rspamd_stat_update_handler, stat_update_time, stat_update_time); ev_timer_start(event_loop, &stat_ev); rspamd_check_core_limits(rspamd_main); rspamd_mempool_lock_mutex(rspamd_main->start_mtx); spawn_workers(rspamd_main, event_loop); rspamd_mempool_unlock_mutex(rspamd_main->start_mtx); rspamd_main->http_ctx = rspamd_http_context_create(rspamd_main->cfg, event_loop, rspamd_main->cfg->ups_ctx); if (control_fd != -1) { msg_info_main("listening for control commands on %s", rspamd_inet_address_to_string(control_addr)); ev_io_init(&control_ev, rspamd_control_handler, control_fd, EV_READ); control_ev.data = rspamd_main; ev_io_start(event_loop, &control_ev); } ev_loop(event_loop, 0); /* Maybe save roll history */ if (rspamd_main->history && rspamd_main->cfg->history_file) { rspamd_roll_history_save(rspamd_main->history, rspamd_main->cfg->history_file); } if (rspamd_main->cfg->cache) { rspamd_symcache_save(rspamd_main->cfg->cache); } msg_info_main("terminating..."); #ifdef WITH_HYPERSCAN rspamd_hyperscan_cleanup_maybe(); #endif REF_RELEASE(rspamd_main->cfg); rspamd_log_close(rspamd_main->logger); g_hash_table_unref(rspamd_main->spairs); g_hash_table_unref(rspamd_main->workers); rspamd_mempool_delete(rspamd_main->server_pool); if (!skip_pid) { rspamd_pidfile_close(rspamd_main->pfh); } rspamd_unset_crash_handler(rspamd_main); g_free(rspamd_main); ev_unref(event_loop); sqlite3_shutdown(); if (control_addr) { rspamd_inet_address_free(control_addr); } return (res); }