[Feature] Add protection against open files limit and accepting sockets

diff --git a/src/controller.c b/src/controller.c
index 91f4cfed6921e226d12fd5e8b2b8983a78db5bdf..f5efb45356a73af4f3443d32940c6fad18849d9c 100644 (file)
--- a/src/controller.c
+++ b/src/controller.c
@@ -2216,7 +2216,7 @@ rspamd_controller_accept_socket (gint fd, short what, void *arg)
        ctx = worker->ctx;
 
        if ((nfd =
-               rspamd_accept_from_socket (fd, &addr)) == -1) {
+               rspamd_accept_from_socket (fd, &addr, worker->accept_events)) == -1) {
                msg_warn_ctx ("accept failed: %s", strerror (errno));
                return;
        }

diff --git a/src/fuzzy_storage.c b/src/fuzzy_storage.c
index bd888fd3cbec54474483c6117d5a9e127e2c2f6d..d96346ce4bd82cbe305f6e9065186c6e916d33f5 100644 (file)
--- a/src/fuzzy_storage.c
+++ b/src/fuzzy_storage.c
@@ -1161,7 +1161,7 @@ accept_fuzzy_mirror_socket (gint fd, short what, void *arg)
        struct fuzzy_master_update_session *session;
 
        if ((nfd =
-                       rspamd_accept_from_socket (fd, &addr)) == -1) {
+                       rspamd_accept_from_socket (fd, &addr, worker->accept_events)) == -1) {
                msg_warn ("accept failed: %s", strerror (errno));
                return;
        }
@@ -2006,7 +2006,7 @@ fuzzy_peer_rep (struct rspamd_worker *worker,
        struct rspamd_fuzzy_storage_ctx *ctx = ud;
        GList *cur;
        struct rspamd_worker_listen_socket *ls;
-       struct event *accept_event;
+       struct event *accept_events;
        gdouble next_check;
 
        ctx->peer_fd = rep_fd;
@@ -2026,23 +2026,23 @@ fuzzy_peer_rep (struct rspamd_worker *worker,
 
                if (ls->fd != -1) {
                        if (ls->type == RSPAMD_WORKER_SOCKET_UDP) {
-                               accept_event = g_slice_alloc0 (sizeof (struct event));
-                               event_set (accept_event, ls->fd, EV_READ | EV_PERSIST,
+                               accept_events = g_slice_alloc0 (sizeof (struct event) * 2);
+                               event_set (&accept_events[0], ls->fd, EV_READ | EV_PERSIST,
                                                accept_fuzzy_socket, worker);
-                               event_base_set (ctx->ev_base, accept_event);
-                               event_add (accept_event, NULL);
+                               event_base_set (ctx->ev_base, &accept_events[0]);
+                               event_add (&accept_events[0], NULL);
                                worker->accept_events = g_list_prepend (worker->accept_events,
-                                               accept_event);
+                                               accept_events);
                        }
                        else if (worker->index == 0) {
                                /* We allow TCP listeners only for a update worker */
-                               accept_event = g_slice_alloc0 (sizeof (struct event));
-                               event_set (accept_event, ls->fd, EV_READ | EV_PERSIST,
+                               accept_events = g_slice_alloc0 (sizeof (struct event) * 2);
+                               event_set (&accept_events[0], ls->fd, EV_READ | EV_PERSIST,
                                                accept_fuzzy_mirror_socket, worker);
-                               event_base_set (ctx->ev_base, accept_event);
-                               event_add (accept_event, NULL);
+                               event_base_set (ctx->ev_base, &accept_events[0]);
+                               event_add (&accept_events[0], NULL);
                                worker->accept_events = g_list_prepend (worker->accept_events,
-                                               accept_event);
+                                               accept_events);
                        }
                }
 

diff --git a/src/libserver/worker_util.c b/src/libserver/worker_util.c
index d3e296d6df3eceeca4860f215a11ed474f43577d..b50b0384f17453a4a66e33e6dd0c78c36d879b44 100644 (file)
--- a/src/libserver/worker_util.c
+++ b/src/libserver/worker_util.c
@@ -248,7 +248,7 @@ rspamd_prepare_worker (struct rspamd_worker *worker, const char *name,
        void (*accept_handler)(int, short, void *))
 {
        struct event_base *ev_base;
-       struct event *accept_event;
+       struct event *accept_events;
        GList *cur;
        struct rspamd_worker_listen_socket *ls;
 
@@ -276,13 +276,13 @@ rspamd_prepare_worker (struct rspamd_worker *worker, const char *name,
                        ls = cur->data;
 
                        if (ls->fd != -1) {
-                               accept_event = g_slice_alloc0 (sizeof (struct event));
-                               event_set (accept_event, ls->fd, EV_READ | EV_PERSIST,
+                               accept_events = g_slice_alloc0 (sizeof (struct event) * 2);
+                               event_set (&accept_events[0], ls->fd, EV_READ | EV_PERSIST,
                                                accept_handler, worker);
-                               event_base_set (ev_base, accept_event);
-                               event_add (accept_event, NULL);
+                               event_base_set (ev_base, &accept_events[0]);
+                               event_add (&accept_events[0], NULL);
                                worker->accept_events = g_list_prepend (worker->accept_events,
-                                               accept_event);
+                                               accept_events);
                        }
 
                        cur = g_list_next (cur);
@@ -296,7 +296,7 @@ void
 rspamd_worker_stop_accept (struct rspamd_worker *worker)
 {
        GList *cur;
-       struct event *event;
+       struct event *events;
        GHashTableIter it;
        struct rspamd_worker_signal_handler *sigh;
        gpointer k, v;
@@ -305,10 +305,11 @@ rspamd_worker_stop_accept (struct rspamd_worker *worker)
        /* Remove all events */
        cur = worker->accept_events;
        while (cur) {
-               event = cur->data;
-               event_del (event);
+               events = cur->data;
+               event_del (&events[0]);
+               event_del (&events[1]);
                cur = g_list_next (cur);
-               g_slice_free1 (sizeof (struct event), event);
+               g_slice_free1 (sizeof (struct event) * 2, events);
        }
 
        if (worker->accept_events != NULL) {

diff --git a/src/libutil/addr.c b/src/libutil/addr.c
index 5c5a51e130805164efd55ea7d555685cb6b7005b..3fee0b4bc2ffbc2393ddb3299c74621e64c03a08 100644 (file)
--- a/src/libutil/addr.c
+++ b/src/libutil/addr.c
@@ -199,8 +199,41 @@ rspamd_ip_is_valid (const rspamd_inet_addr_t *addr)
        return ret;
 }
 
+static void
+rspamd_enable_accept_event (gint fd, short what, gpointer d)
+{
+       struct event *events = d;
+
+       event_del (&events[1]);
+       event_add (&events[0], NULL);
+}
+
+static void
+rspamd_disable_accept_events (gint sock, GList *accept_events)
+{
+       GList *cur;
+       struct event *events;
+       const gdouble throttling = 0.5;
+       struct timeval tv;
+       struct event_base *ev_base;
+
+       double_to_tv (throttling, &tv);
+
+       for (cur = accept_events; cur != NULL; cur = g_list_next (cur)) {
+               events = cur->data;
+
+               ev_base = event_get_base (&events[0]);
+               event_del (&events[0]);
+               event_set (&events[1], sock, EV_TIMEOUT, rspamd_enable_accept_event,
+                               events);
+               event_base_set (ev_base, &events[1]);
+               event_add (&events[1], &tv);
+       }
+}
+
 gint
-rspamd_accept_from_socket (gint sock, rspamd_inet_addr_t **target)
+rspamd_accept_from_socket (gint sock, rspamd_inet_addr_t **target,
+               GList *accept_events)
 {
        gint nfd, serrno;
        union sa_union su;
@@ -215,6 +248,13 @@ rspamd_accept_from_socket (gint sock, rspamd_inet_addr_t **target)
                if (errno == EAGAIN || errno == EINTR || errno == EWOULDBLOCK) {
                        return 0;
                }
+               else if (errno == EMFILE || errno == ENFILE) {
+                       /* Temporary disable accept event */
+                       rspamd_disable_accept_events (sock, accept_events);
+
+                       return 0;
+               }
+
                return -1;
        }
 

diff --git a/src/libutil/addr.h b/src/libutil/addr.h
index bb9fd25736028d2bf52ae646a83c1c6b4ef14ad2..200543d6f76f04fa15f9df6aa2443caba629c9a2 100644 (file)
--- a/src/libutil/addr.h
+++ b/src/libutil/addr.h
@@ -193,10 +193,12 @@ gboolean rspamd_ip_is_valid (const rspamd_inet_addr_t *addr);
 /**
  * Accept from listening socket filling addr structure
  * @param sock listening socket
- * @param addr allocated inet addr structur
+ * @param addr allocated inet addr structure
+ * @param accept_events events for accepting new sockets
  * @return
  */
-gint rspamd_accept_from_socket (gint sock, rspamd_inet_addr_t **addr);
+gint rspamd_accept_from_socket (gint sock, rspamd_inet_addr_t **addr,
+               GList *accept_events);
 
 /**
  * Parse host[:port[:priority]] line

diff --git a/src/lua_worker.c b/src/lua_worker.c
index df6970efac302556393eff84a8e9eec91419e638..b74b8d42222143afa2ee8c12615906d7d701bf2c 100644 (file)
--- a/src/lua_worker.c
+++ b/src/lua_worker.c
@@ -261,7 +261,7 @@ lua_accept_socket (gint fd, short what, void *arg)
        L = ctx->L;
 
        if ((nfd =
-               rspamd_accept_from_socket (fd, &addr)) == -1) {
+               rspamd_accept_from_socket (fd, &addr, worker->accept_events)) == -1) {
                msg_warn ("accept failed: %s", strerror (errno));
                return;
        }

diff --git a/src/rspamadm/lua_repl.c b/src/rspamadm/lua_repl.c
index f0d337ad9a1c61fe9a137db99fc4b6e354348b38..f96aa289009876d941dc11ee5c2aa85088f7ce91 100644 (file)
--- a/src/rspamadm/lua_repl.c
+++ b/src/rspamadm/lua_repl.c
@@ -482,7 +482,7 @@ rspamadm_lua_accept_cb (gint fd, short what, void *arg)
        gint nfd;
 
        if ((nfd =
-                       rspamd_accept_from_socket (fd, &addr)) == -1) {
+                       rspamd_accept_from_socket (fd, &addr, NULL)) == -1) {
                rspamd_fprintf (stderr, "accept failed: %s", strerror (errno));
                return;
        }

diff --git a/src/rspamd.c b/src/rspamd.c
index b6b5a271c70cd2ffaad344b26e66f7177ad695a4..922327f3840916e17a4e86866ef70fabf10e32bd 100644 (file)
--- a/src/rspamd.c
+++ b/src/rspamd.c
@@ -959,7 +959,7 @@ rspamd_control_handler (gint fd, short what, gpointer arg)
        gint nfd;
 
        if ((nfd =
-                                rspamd_accept_from_socket (fd, &addr)) == -1) {
+                                rspamd_accept_from_socket (fd, &addr, NULL)) == -1) {
                msg_warn_main ("accept failed: %s", strerror (errno));
                return;
        }

diff --git a/src/rspamd_proxy.c b/src/rspamd_proxy.c
index 7ca0cc51d70ee9356a042a34ade65fd3454c7516..d2ac90aa08cc63f296921ddd52275c2d0ac0ce1f 100644 (file)
--- a/src/rspamd_proxy.c
+++ b/src/rspamd_proxy.c
@@ -1232,7 +1232,7 @@ proxy_accept_socket (gint fd, short what, void *arg)
        ctx = worker->ctx;
 
        if ((nfd =
-               rspamd_accept_from_socket (fd, &addr)) == -1) {
+               rspamd_accept_from_socket (fd, &addr, worker->accept_events)) == -1) {
                msg_warn ("accept failed: %s", strerror (errno));
                return;
        }

diff --git a/src/smtp_proxy.c b/src/smtp_proxy.c
index 6b0a4fe2e898a3b00a883647736007a528144884..8eebc2c8639b2da13425653000e272302fd8f50b 100644 (file)
--- a/src/smtp_proxy.c
+++ b/src/smtp_proxy.c
@@ -902,7 +902,7 @@ accept_socket (gint fd, short what, void *arg)
        ctx = worker->ctx;
 
        if ((nfd =
-               rspamd_accept_from_socket (fd, &addr)) == -1) {
+               rspamd_accept_from_socket (fd, &addr, worker->accept_events)) == -1) {
                msg_warn ("accept failed: %s", strerror (errno));
                return;
        }

diff --git a/src/worker.c b/src/worker.c
index ac104f7f0f8ba659841c99290b67b80de59fcafb..81c5b1c63e3b224d271e9861c9c16659fe295c9d 100644 (file)
--- a/src/worker.c
+++ b/src/worker.c
@@ -266,7 +266,7 @@ accept_socket (gint fd, short what, void *arg)
        }
 
        if ((nfd =
-               rspamd_accept_from_socket (fd, &addr)) == -1) {
+               rspamd_accept_from_socket (fd, &addr, worker->accept_events)) == -1) {
                msg_warn_ctx ("accept failed: %s", strerror (errno));
                return;
        }

diff --git a/test/rspamd_http_test.c b/test/rspamd_http_test.c
index 428c510c41a171ce388ff5d6d4c2c3c66d83cb19..88ac4867687d67d8c554d1c44fdd248a4d5db51f 100644 (file)
--- a/test/rspamd_http_test.c
+++ b/test/rspamd_http_test.c
@@ -55,7 +55,7 @@ rspamd_server_accept (gint fd, short what, void *arg)
        gint nfd;
 
        if ((nfd =
-                       rspamd_accept_from_socket (fd, &addr)) == -1) {
+                       rspamd_accept_from_socket (fd, &addr, NULL)) == -1) {
                msg_warn ("accept failed: %s", strerror (errno));
                return;
        }

diff --git a/utils/rspamd_http_server.c b/utils/rspamd_http_server.c
index ad01085dfc2bc9a0023ceffdb91f8fb2d6d46d55..9af96fad97dfcbf7519707c30f7613a70825e4b0 100644 (file)
--- a/utils/rspamd_http_server.c
+++ b/utils/rspamd_http_server.c
@@ -134,7 +134,7 @@ rspamd_server_accept (gint fd, short what, void *arg)
 
        do {
                if ((nfd =
-                                        rspamd_accept_from_socket (fd, &addr)) == -1) {
+                                        rspamd_accept_from_socket (fd, &addr, NULL)) == -1) {
                        rspamd_fprintf (stderr, "accept failed: %s", strerror (errno));
                        return;
                }