lib-master, master: Fix inet_listener_reuse_port=yes handling

This only works properly if the listener sockets don't change and
there is always a process listening on each of the sockets.

Now master process creates the reuse_port=yes listeners for all of the
processes at startup (up to process_limit). Each child process inherits
exactly one unique listener socket based on its assigned process index.

Since with reuse_port=yes connections are assigned to a specific socket,
processes can no longer rely on other processes picking up connections when
they are full. Instead, each process is now responsible for rejecting
connections when they reach client_limit.

diff --git a/src/lib-master/master-interface.h b/src/lib-master/master-interface.h
index e420157130527a7bb4f8b0a69f610efa1b0bce22..fd3b6be739e13682a94a2723efc6d0c380661894 100644 (file)
--- a/src/lib-master/master-interface.h
+++ b/src/lib-master/master-interface.h
@@ -65,6 +65,9 @@ enum master_login_state {
    timeout in seconds. */
 #define MASTER_SERVICE_IDLE_KILL_INTERVAL_ENV "IDLE_KILL_INTERVAL"
 
+/* getenv(MASTER_REUSE_PORT_ENV) is non-NULL if service_reuse_port=yes */
+#define MASTER_REUSE_PORT_ENV "REUSE_PORT"
+
 /* getenv(MASTER_CONFIG_FILE_ENV) provides path to configuration file. */
 #define MASTER_CONFIG_FILE_ENV "CONFIG_FILE"
 

diff --git a/src/lib-master/master-service-private.h b/src/lib-master/master-service-private.h
index 7385ab353438f77de4cbca765cf583dbbcbb18c5..df67ac7181eb5adf24b12327afa8e12912fbfad4 100644 (file)
--- a/src/lib-master/master-service-private.h
+++ b/src/lib-master/master-service-private.h
@@ -104,6 +104,7 @@ struct master_service {
        bool call_avail_overflow:1;
        bool config_path_changed_with_param:1;
        bool have_admin_sockets:1;
+       bool reuse_port:1;
        bool want_ssl_server:1;
        bool config_path_from_master:1;
        bool log_initialized:1;

diff --git a/src/lib-master/master-service.c b/src/lib-master/master-service.c
index 73906922a7bdf6a6fbd99ed334b2ceddcff1d929..2d117fed6ee10d8318391c3d2b4dca03dc446c81 100644 (file)
--- a/src/lib-master/master-service.c
+++ b/src/lib-master/master-service.c
@@ -570,6 +570,7 @@ master_service_init(const char *name, enum master_service_flags flags,
        } else {
                service->version_string = PACKAGE_VERSION;
        }
+       service->reuse_port = getenv(MASTER_REUSE_PORT_ENV) != NULL;
 
        /* Load the SSL module if we already know it is necessary. It can also
           get loaded later on-demand. */
@@ -1798,12 +1799,20 @@ static bool master_service_full(struct master_service *service)
        struct timeval created;
 
        /* This process can't handle any more connections. */
-       if (!service->call_avail_overflow ||
-           service->avail_overflow_callback == NULL)
+       if (service->avail_overflow_callback == NULL)
                return TRUE;
+       if (!service->call_avail_overflow && !service->reuse_port) {
+               /* This process is full, but sibling processes aren't. Another
+                  process will pick up this connection. Note that with
+                  reuse_port=yes the connection is specifically assigned to
+                  this process, so we are responsible for accepting or
+                  rejecting it. */
+               return TRUE;
+       }
 
-       /* Master has notified us that all processes are full, and
-          we have the ability to kill old connections. */
+       /* Master has notified us that all processes are full (or with
+          reuse_port=yes this process is full), and we have the ability to
+          kill old connections. */
        if (service->total_available_count > 1) {
                /* This process can still create multiple concurrent
                   clients if we just kill some of the existing ones.
@@ -1895,6 +1904,15 @@ static void master_service_listen(struct master_service_listener *l)
 
        if (service->master_status.available_count == 0 && !master_admin_conn) {
                if (master_service_full(service)) {
+                       if (service->reuse_port) {
+                               /* With reuse_port the master can't drop
+                                  the connections for us. We must do it
+                                  ourselves. */
+                               int fd = net_accept(l->fd, NULL, NULL);
+                               if (fd != -1)
+                                       i_close_fd(&fd);
+                               return;
+                       }
                        /* Stop the listener until a client has disconnected or
                           overflow callback has killed one. */
                        master_service_io_listeners_remove(service);

diff --git a/src/master/master-settings.c b/src/master/master-settings.c
index 67034395b061f10903db66bfba3e7d4fd758aca1..7a2bb5fa3b4a4dedf2e25656ee9f31e78dee7738 100644 (file)
--- a/src/master/master-settings.c
+++ b/src/master/master-settings.c
@@ -776,31 +776,12 @@ master_settings_ext_check(struct event *event, void *_set,
                        return FALSE;
                }
 
-               if (array_is_created(&service->parsed_inet_listeners)) {
-                       struct inet_listener_settings *l;
-                       bool seen_reuse_port = FALSE;
-
-                       array_foreach_elem(&service->parsed_inet_listeners, l) {
-                               if (l->port == 0)
-                                       continue;
-
-                               if (l->reuse_port)
-                                       seen_reuse_port = TRUE;
-                               else if (seen_reuse_port) {
-                                       *error_r = t_strdup_printf("service(%s): "
-                                               "All or none of the inet_listeners must have reuse_port=yes "
-                                               "(missing for inet_listener %s)",
-                                               service->name, l->name);
-                                       return FALSE;
-                               }
-                               if (l->reuse_port &&
-                                   service->process_min_avail != service->process_limit) {
-                                       *error_r = t_strdup_printf("service(%s): "
-                                               "process_min_avail must be equal to process_limit "
-                                               "when using reuse_port=yes", service->name);
-                                       return FALSE;
-                               }
-                       }
+               if (service->reuse_port &&
+                   service->process_min_avail != service->process_limit) {
+                       *error_r = t_strdup_printf("service(%s): "
+                               "process_min_avail must be equal to process_limit when using service_reuse_port=yes",
+                               service->name);
+                       return FALSE;
                }
 
 #ifdef CONFIG_BINARY

diff --git a/src/master/service-listen.c b/src/master/service-listen.c
index b8dd596d7eb348f41792e654f1532928c5ae58c9..74a60337ea8ec7e1a699b11fee0c1c33b8847b87 100644 (file)
--- a/src/master/service-listen.c
+++ b/src/master/service-listen.c
@@ -424,7 +424,7 @@ static bool listener_equals(const struct service_listener *l1,
                        return FALSE;
                if (l1->set.inetset.set->port != l2->set.inetset.set->port)
                        return FALSE;
-               return TRUE;
+               return l1->reuse_port_process_index == l2->reuse_port_process_index;
        }
        return FALSE;
 }

diff --git a/src/master/service-monitor.c b/src/master/service-monitor.c
index 84a6497f727de9550fdb2a7884b91d4b6dc05886..a7b60ac20c60600c04d86e246ef8662fa9eaa0a9 100644 (file)
--- a/src/master/service-monitor.c
+++ b/src/master/service-monitor.c
@@ -135,6 +135,15 @@ static void service_status_more(struct service_process *process,
            service_active_process_count(service) >= service->process_limit)
                service_login_notify(service, TRUE);
 
+       if (service->set->reuse_port &&
+           service->last_drop_warning +
+           SERVICE_DROP_WARN_INTERVAL_SECS <= ioloop_time) {
+               service->last_drop_warning = ioloop_time;
+               e_warning(service->event,
+                         "client_limit (%u) reached, client connections are being dropped (pid=%s, reuse_port=yes)",
+                         service->client_limit, dec2str(process->pid));
+       }
+
        /* we may need to start more */
        service_monitor_start_extra_avail(service);
        service_monitor_listen_start(service);
@@ -514,7 +523,7 @@ static void service_monitor_listen_start_force(struct service *service)
        timeout_remove(&service->to_drop_warning);
 
        array_foreach_elem(&service->listeners, l) {
-               if (l->io == NULL && l->fd != -1)
+               if (l->io == NULL && l->fd != -1 && !service->set->reuse_port)
                        l->io = io_add(l->fd, IO_READ, service_accept, l);
        }
 }

diff --git a/src/master/service-process.c b/src/master/service-process.c
index cd6b5a4ce363f64c93f62c322b6b5dbb4137f049..f91ce313279151ff8417b97d40782a7d82df5c05 100644 (file)
--- a/src/master/service-process.c
+++ b/src/master/service-process.c
@@ -34,24 +34,6 @@
 #include <signal.h>
 #include <sys/wait.h>
 
-static void service_reopen_inet_listeners(struct service *service)
-{
-       struct service_listener *const *listeners;
-       unsigned int i, count;
-       int old_fd;
-
-       listeners = array_get(&service->listeners, &count);
-       for (i = 0; i < count; i++) {
-               if (!service->set->reuse_port || listeners[i]->fd == -1)
-                       continue;
-
-               old_fd = listeners[i]->fd;
-               listeners[i]->fd = -1;
-               if (service_listener_listen(listeners[i]) < 0)
-                       listeners[i]->fd = old_fd;
-       }
-}
-
 static int
 service_unix_pid_listener_get_path(struct service_listener *l, pid_t pid,
                                   string_t *path, const char **error_r)
@@ -69,7 +51,8 @@ service_unix_pid_listener_get_path(struct service_listener *l, pid_t pid,
 }
 
 static void
-service_dup_fds(struct service *service, int accepted_fd,
+service_dup_fds(struct service *service, unsigned int process_index,
+               int accepted_fd,
                const struct service_listener *accepted_listener,
                int *accepted_listener_fd_r)
 {
@@ -147,6 +130,13 @@ service_dup_fds(struct service *service, int accepted_fd,
                                }
                        }
 
+                       /* The listeners array contains reuse_port=yes listeners
+                          for every process. Here we filter out the listeners
+                          not intended for this process. */
+                       if (service->set->reuse_port &&
+                           listeners[i]->reuse_port_process_index != process_index)
+                               continue;
+
                        if (listeners[i] == accepted_listener)
                                *accepted_listener_fd_r = fd;
                        dup2_append(&dups, listeners[i]->fd, fd++);
@@ -327,6 +317,8 @@ service_process_setup_environment(struct service *service, unsigned int uid,
                env_put(MASTER_SERVICE_COUNT_ENV,
                        dec2str(service->set->restart_request_count));
        }
+       if (service->set->reuse_port)
+               env_put(MASTER_REUSE_PORT_ENV, "1");
        env_put(MASTER_UID_ENV, dec2str(uid));
        env_put(MY_HOSTNAME_ENV, my_hostname);
        env_put(MY_HOSTDOMAIN_ENV, hostdomain);
@@ -398,6 +390,29 @@ service_process_create(struct service *service, int accepted_fd,
           future lookups. */
        hostdomain = my_hostdomain();
 
+       unsigned int process_index = 0;
+       if (service->set->reuse_port) {
+               /* Figure out the process index number. Do this by scanning
+                  all the existing processes for the service and finding the
+                  first nonexistent index number. Note that retired processes
+                  are no longer listening, so their index must be reused. */
+               bool *seen_index = t_new(bool, service->process_limit);
+               struct service_process *p;
+               for (p = service->busy_processes; p != NULL; p = p->next) {
+                       if (p->index < service->process_limit && !p->retired)
+                               seen_index[p->index] = TRUE;
+               }
+               for (p = service->idle_processes_head; p != NULL; p = p->next) {
+                       if (p->index < service->process_limit)
+                               seen_index[p->index] = TRUE;
+               }
+               for (process_index = 0; process_index < service->process_limit; process_index++) {
+                       if (!seen_index[process_index])
+                               break;
+               }
+               i_assert(process_index < service->process_limit);
+       }
+
        if (service->type == SERVICE_TYPE_ANVIL &&
            service_anvil_global->pid != 0) {
                pid = service_anvil_global->pid;
@@ -427,8 +442,7 @@ service_process_create(struct service *service, int accepted_fd,
                /* child */
                int accepted_listener_fd;
                service_process_setup_environment(service, uid, hostdomain);
-               service_reopen_inet_listeners(service);
-               service_dup_fds(service, accepted_fd, accepted_listener,
+               service_dup_fds(service, process_index, accepted_fd, accepted_listener,
                                &accepted_listener_fd);
                if (accepted_fd != -1) {
                        i_assert(accepted_listener_fd > 0);
@@ -445,6 +459,7 @@ service_process_create(struct service *service, int accepted_fd,
        process->refcount = 1;
        process->pid = pid;
        process->uid = uid;
+       process->index = process_index;
        process->create_time = ioloop_time;
        if (process_forked) {
                process->to_status =

diff --git a/src/master/service-process.h b/src/master/service-process.h
index bc1a0c2f8ec09b784b90aa190843a7943e8513d9..95b067d2d7806241151417907a486525b4fde7a1 100644 (file)
--- a/src/master/service-process.h
+++ b/src/master/service-process.h
@@ -19,6 +19,11 @@ struct service_process {
           smaller than the correct value. */
        unsigned int total_count;
 
+       /* Process index number. This is set only for services with
+          inet_listener_reuse_port=yes listeners. See
+          service_listener.reuse_port_process_index for how this is used. */
+       unsigned int index;
+
        /* Timestamp when the process was created */
        time_t create_time;
        /* Time when process started idling, or 0 if we're not idling. This is

diff --git a/src/master/service.c b/src/master/service.c
index cd58f03789012edcc9fd2f0bfc45cada2f7775ab..4fe6299fa60f1824c9605b3f4a55c88c8785e544 100644 (file)
--- a/src/master/service.c
+++ b/src/master/service.c
@@ -156,9 +156,19 @@ service_create_inet_listeners(struct service *service,
                        return -1;
 
                for (i = 0; i < ips_count; i++) {
-                       l = service_create_one_inet_listener(service, set,
-                                                            address, &ips[i]);
-                       array_push_back(&service->listeners, &l);
+                       /* reuse_port=yes listeners create all of the processes'
+                          listeners at startup. */
+                       unsigned int j, count;
+                       if (!service->set->reuse_port)
+                               count = 1;
+                       else
+                               count = service->process_limit;
+                       for (j = 0; j < count; j++) {
+                               l = service_create_one_inet_listener(service, set,
+                                                                    address, &ips[i]);
+                               l->reuse_port_process_index = j;
+                               array_push_back(&service->listeners, &l);
+                       }
                }
                service->have_inet_listeners = TRUE;
        }
@@ -577,6 +587,11 @@ void service_login_notify(struct service *service, bool all_processes_full)
        if (service->last_login_full_notify == all_processes_full ||
            service->login_notify_fd == -1)
                return;
+       if (service->set->reuse_port) {
+               /* With reuse_port=yes the processes don't care about sibling
+                  processes' state. */
+               return;
+       }
 
        /* change the state always immediately. it's cheap. */
        service->last_login_full_notify = all_processes_full;

diff --git a/src/master/service.h b/src/master/service.h
index 6582e6504eeda493359950336095eca0f221c653..f089d7be1f95abebfe1436eff53ea5e976140769 100644 (file)
--- a/src/master/service.h
+++ b/src/master/service.h
@@ -39,6 +39,14 @@ struct service_listener {
                        struct ip_addr ip;
                } inetset;
        } set;
+
+       /* This listener is used only by the process with this index number.
+          Each process gets their own reuse_port listener. This index number
+          is between 0..(process_limit-1) and valid only for that process
+          index. The index is assigned while at the process creation order.
+          If a process exits or retires, it is replaced by a new process with
+          the same index number. */
+       unsigned int reuse_port_process_index;
 };
 
 struct service {