Mercurial > hg > nginx-quic
view src/os/unix/ngx_process.c @ 4667:d05ab8793a69 stable-1.2
Merge of r4622, r4623: balancing changes.
*) Upstream: smooth weighted round-robin balancing.
For edge case weights like { 5, 1, 1 } we now produce { a, a, b, a, c, a, a }
sequence instead of { c, b, a, a, a, a, a } produced previously.
Algorithm is as follows: on each peer selection we increase current_weight
of each eligible peer by its weight, select peer with greatest current_weight
and reduce its current_weight by total number of weight points distributed
among peers.
In case of { 5, 1, 1 } weights this gives the following sequence of
current_weight's:
a b c
0 0 0 (initial state)
5 1 1 (a selected)
-2 1 1
3 2 2 (a selected)
-4 2 2
1 3 3 (b selected)
1 -4 3
6 -3 4 (a selected)
-1 -3 4
4 -2 5 (c selected)
4 -2 -2
9 -1 -1 (a selected)
2 -1 -1
7 0 0 (a selected)
0 0 0
To preserve weight reduction in case of failures the effective_weight
variable was introduced, which usually matches peer's weight, but is
reduced temporarily on peer failures.
This change also fixes loop with backup servers and proxy_next_upstream
http_404 (ticket #47), and skipping alive upstreams in some cases if there
are multiple dead ones (ticket #64).
*) Upstream: fixed ip_hash rebalancing with the "down" flag.
Due to weight being set to 0 for down peers, order of peers after sorting
wasn't the same as without the "down" flag (with down peers at the end),
resulting in client rebalancing for clients on other servers. The only
rebalancing which should happen after adding "down" to a server is one
for clients on the server.
The problem was introduced in r1377 (which fixed endless loop by setting
weight to 0 for down servers). The loop is no longer possible with new
smooth algorithm, so preserving original weight is safe.
author | Maxim Dounin <mdounin@mdounin.ru> |
---|---|
date | Mon, 04 Jun 2012 11:21:58 +0000 |
parents | 14411ee4d89f |
children | a50ed82f19e4 |
line wrap: on
line source
/* * Copyright (C) Igor Sysoev * Copyright (C) Nginx, Inc. */ #include <ngx_config.h> #include <ngx_core.h> #include <ngx_event.h> #include <ngx_channel.h> typedef struct { int signo; char *signame; char *name; void (*handler)(int signo); } ngx_signal_t; static void ngx_execute_proc(ngx_cycle_t *cycle, void *data); static void ngx_signal_handler(int signo); static void ngx_process_get_status(void); static void ngx_unlock_mutexes(ngx_pid_t pid); int ngx_argc; char **ngx_argv; char **ngx_os_argv; ngx_int_t ngx_process_slot; ngx_socket_t ngx_channel; ngx_int_t ngx_last_process; ngx_process_t ngx_processes[NGX_MAX_PROCESSES]; ngx_signal_t signals[] = { { ngx_signal_value(NGX_RECONFIGURE_SIGNAL), "SIG" ngx_value(NGX_RECONFIGURE_SIGNAL), "reload", ngx_signal_handler }, { ngx_signal_value(NGX_REOPEN_SIGNAL), "SIG" ngx_value(NGX_REOPEN_SIGNAL), "reopen", ngx_signal_handler }, { ngx_signal_value(NGX_NOACCEPT_SIGNAL), "SIG" ngx_value(NGX_NOACCEPT_SIGNAL), "", ngx_signal_handler }, { ngx_signal_value(NGX_TERMINATE_SIGNAL), "SIG" ngx_value(NGX_TERMINATE_SIGNAL), "stop", ngx_signal_handler }, { ngx_signal_value(NGX_SHUTDOWN_SIGNAL), "SIG" ngx_value(NGX_SHUTDOWN_SIGNAL), "quit", ngx_signal_handler }, { ngx_signal_value(NGX_CHANGEBIN_SIGNAL), "SIG" ngx_value(NGX_CHANGEBIN_SIGNAL), "", ngx_signal_handler }, { SIGALRM, "SIGALRM", "", ngx_signal_handler }, { SIGINT, "SIGINT", "", ngx_signal_handler }, { SIGIO, "SIGIO", "", ngx_signal_handler }, { SIGCHLD, "SIGCHLD", "", ngx_signal_handler }, { SIGSYS, "SIGSYS, SIG_IGN", "", SIG_IGN }, { SIGPIPE, "SIGPIPE, SIG_IGN", "", SIG_IGN }, { 0, NULL, "", NULL } }; ngx_pid_t ngx_spawn_process(ngx_cycle_t *cycle, ngx_spawn_proc_pt proc, void *data, char *name, ngx_int_t respawn) { u_long on; ngx_pid_t pid; ngx_int_t s; if (respawn >= 0) { s = respawn; } else { for (s = 0; s < ngx_last_process; s++) { if (ngx_processes[s].pid == -1) { break; } } if (s == NGX_MAX_PROCESSES) { ngx_log_error(NGX_LOG_ALERT, cycle->log, 0, "no more than %d processes can be spawned", NGX_MAX_PROCESSES); return NGX_INVALID_PID; } } if (respawn != NGX_PROCESS_DETACHED) { /* Solaris 9 still has no AF_LOCAL */ if (socketpair(AF_UNIX, SOCK_STREAM, 0, ngx_processes[s].channel) == -1) { ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "socketpair() failed while spawning \"%s\"", name); return NGX_INVALID_PID; } ngx_log_debug2(NGX_LOG_DEBUG_CORE, cycle->log, 0, "channel %d:%d", ngx_processes[s].channel[0], ngx_processes[s].channel[1]); if (ngx_nonblocking(ngx_processes[s].channel[0]) == -1) { ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, ngx_nonblocking_n " failed while spawning \"%s\"", name); ngx_close_channel(ngx_processes[s].channel, cycle->log); return NGX_INVALID_PID; } if (ngx_nonblocking(ngx_processes[s].channel[1]) == -1) { ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, ngx_nonblocking_n " failed while spawning \"%s\"", name); ngx_close_channel(ngx_processes[s].channel, cycle->log); return NGX_INVALID_PID; } on = 1; if (ioctl(ngx_processes[s].channel[0], FIOASYNC, &on) == -1) { ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "ioctl(FIOASYNC) failed while spawning \"%s\"", name); ngx_close_channel(ngx_processes[s].channel, cycle->log); return NGX_INVALID_PID; } if (fcntl(ngx_processes[s].channel[0], F_SETOWN, ngx_pid) == -1) { ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "fcntl(F_SETOWN) failed while spawning \"%s\"", name); ngx_close_channel(ngx_processes[s].channel, cycle->log); return NGX_INVALID_PID; } if (fcntl(ngx_processes[s].channel[0], F_SETFD, FD_CLOEXEC) == -1) { ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "fcntl(FD_CLOEXEC) failed while spawning \"%s\"", name); ngx_close_channel(ngx_processes[s].channel, cycle->log); return NGX_INVALID_PID; } if (fcntl(ngx_processes[s].channel[1], F_SETFD, FD_CLOEXEC) == -1) { ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "fcntl(FD_CLOEXEC) failed while spawning \"%s\"", name); ngx_close_channel(ngx_processes[s].channel, cycle->log); return NGX_INVALID_PID; } ngx_channel = ngx_processes[s].channel[1]; } else { ngx_processes[s].channel[0] = -1; ngx_processes[s].channel[1] = -1; } ngx_process_slot = s; pid = fork(); switch (pid) { case -1: ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "fork() failed while spawning \"%s\"", name); ngx_close_channel(ngx_processes[s].channel, cycle->log); return NGX_INVALID_PID; case 0: ngx_pid = ngx_getpid(); proc(cycle, data); break; default: break; } ngx_log_error(NGX_LOG_NOTICE, cycle->log, 0, "start %s %P", name, pid); ngx_processes[s].pid = pid; ngx_processes[s].exited = 0; if (respawn >= 0) { return pid; } ngx_processes[s].proc = proc; ngx_processes[s].data = data; ngx_processes[s].name = name; ngx_processes[s].exiting = 0; switch (respawn) { case NGX_PROCESS_NORESPAWN: ngx_processes[s].respawn = 0; ngx_processes[s].just_spawn = 0; ngx_processes[s].detached = 0; break; case NGX_PROCESS_JUST_SPAWN: ngx_processes[s].respawn = 0; ngx_processes[s].just_spawn = 1; ngx_processes[s].detached = 0; break; case NGX_PROCESS_RESPAWN: ngx_processes[s].respawn = 1; ngx_processes[s].just_spawn = 0; ngx_processes[s].detached = 0; break; case NGX_PROCESS_JUST_RESPAWN: ngx_processes[s].respawn = 1; ngx_processes[s].just_spawn = 1; ngx_processes[s].detached = 0; break; case NGX_PROCESS_DETACHED: ngx_processes[s].respawn = 0; ngx_processes[s].just_spawn = 0; ngx_processes[s].detached = 1; break; } if (s == ngx_last_process) { ngx_last_process++; } return pid; } ngx_pid_t ngx_execute(ngx_cycle_t *cycle, ngx_exec_ctx_t *ctx) { return ngx_spawn_process(cycle, ngx_execute_proc, ctx, ctx->name, NGX_PROCESS_DETACHED); } static void ngx_execute_proc(ngx_cycle_t *cycle, void *data) { ngx_exec_ctx_t *ctx = data; if (execve(ctx->path, ctx->argv, ctx->envp) == -1) { ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "execve() failed while executing %s \"%s\"", ctx->name, ctx->path); } exit(1); } ngx_int_t ngx_init_signals(ngx_log_t *log) { ngx_signal_t *sig; struct sigaction sa; for (sig = signals; sig->signo != 0; sig++) { ngx_memzero(&sa, sizeof(struct sigaction)); sa.sa_handler = sig->handler; sigemptyset(&sa.sa_mask); if (sigaction(sig->signo, &sa, NULL) == -1) { ngx_log_error(NGX_LOG_EMERG, log, ngx_errno, "sigaction(%s) failed", sig->signame); return NGX_ERROR; } } return NGX_OK; } void ngx_signal_handler(int signo) { char *action; ngx_int_t ignore; ngx_err_t err; ngx_signal_t *sig; ignore = 0; err = ngx_errno; for (sig = signals; sig->signo != 0; sig++) { if (sig->signo == signo) { break; } } ngx_time_sigsafe_update(); action = ""; switch (ngx_process) { case NGX_PROCESS_MASTER: case NGX_PROCESS_SINGLE: switch (signo) { case ngx_signal_value(NGX_SHUTDOWN_SIGNAL): ngx_quit = 1; action = ", shutting down"; break; case ngx_signal_value(NGX_TERMINATE_SIGNAL): case SIGINT: ngx_terminate = 1; action = ", exiting"; break; case ngx_signal_value(NGX_NOACCEPT_SIGNAL): if (ngx_daemonized) { ngx_noaccept = 1; action = ", stop accepting connections"; } break; case ngx_signal_value(NGX_RECONFIGURE_SIGNAL): ngx_reconfigure = 1; action = ", reconfiguring"; break; case ngx_signal_value(NGX_REOPEN_SIGNAL): ngx_reopen = 1; action = ", reopening logs"; break; case ngx_signal_value(NGX_CHANGEBIN_SIGNAL): if (getppid() > 1 || ngx_new_binary > 0) { /* * Ignore the signal in the new binary if its parent is * not the init process, i.e. the old binary's process * is still running. Or ignore the signal in the old binary's * process if the new binary's process is already running. */ action = ", ignoring"; ignore = 1; break; } ngx_change_binary = 1; action = ", changing binary"; break; case SIGALRM: ngx_sigalrm = 1; break; case SIGIO: ngx_sigio = 1; break; case SIGCHLD: ngx_reap = 1; break; } break; case NGX_PROCESS_WORKER: case NGX_PROCESS_HELPER: switch (signo) { case ngx_signal_value(NGX_NOACCEPT_SIGNAL): if (!ngx_daemonized) { break; } ngx_debug_quit = 1; case ngx_signal_value(NGX_SHUTDOWN_SIGNAL): ngx_quit = 1; action = ", shutting down"; break; case ngx_signal_value(NGX_TERMINATE_SIGNAL): case SIGINT: ngx_terminate = 1; action = ", exiting"; break; case ngx_signal_value(NGX_REOPEN_SIGNAL): ngx_reopen = 1; action = ", reopening logs"; break; case ngx_signal_value(NGX_RECONFIGURE_SIGNAL): case ngx_signal_value(NGX_CHANGEBIN_SIGNAL): case SIGIO: action = ", ignoring"; break; } break; } ngx_log_error(NGX_LOG_NOTICE, ngx_cycle->log, 0, "signal %d (%s) received%s", signo, sig->signame, action); if (ignore) { ngx_log_error(NGX_LOG_CRIT, ngx_cycle->log, 0, "the changing binary signal is ignored: " "you should shutdown or terminate " "before either old or new binary's process"); } if (signo == SIGCHLD) { ngx_process_get_status(); } ngx_set_errno(err); } static void ngx_process_get_status(void) { int status; char *process; ngx_pid_t pid; ngx_err_t err; ngx_int_t i; ngx_uint_t one; one = 0; for ( ;; ) { pid = waitpid(-1, &status, WNOHANG); if (pid == 0) { return; } if (pid == -1) { err = ngx_errno; if (err == NGX_EINTR) { continue; } if (err == NGX_ECHILD && one) { return; } #if (NGX_SOLARIS || NGX_FREEBSD) /* * Solaris always calls the signal handler for each exited process * despite waitpid() may be already called for this process. * * When several processes exit at the same time FreeBSD may * erroneously call the signal handler for exited process * despite waitpid() may be already called for this process. */ if (err == NGX_ECHILD) { ngx_log_error(NGX_LOG_INFO, ngx_cycle->log, err, "waitpid() failed"); return; } #endif ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, err, "waitpid() failed"); return; } one = 1; process = "unknown process"; for (i = 0; i < ngx_last_process; i++) { if (ngx_processes[i].pid == pid) { ngx_processes[i].status = status; ngx_processes[i].exited = 1; process = ngx_processes[i].name; break; } } if (WTERMSIG(status)) { #ifdef WCOREDUMP ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, 0, "%s %P exited on signal %d%s", process, pid, WTERMSIG(status), WCOREDUMP(status) ? " (core dumped)" : ""); #else ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, 0, "%s %P exited on signal %d", process, pid, WTERMSIG(status)); #endif } else { ngx_log_error(NGX_LOG_NOTICE, ngx_cycle->log, 0, "%s %P exited with code %d", process, pid, WEXITSTATUS(status)); } if (WEXITSTATUS(status) == 2 && ngx_processes[i].respawn) { ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, 0, "%s %P exited with fatal code %d " "and cannot be respawned", process, pid, WEXITSTATUS(status)); ngx_processes[i].respawn = 0; } ngx_unlock_mutexes(pid); } } static void ngx_unlock_mutexes(ngx_pid_t pid) { ngx_uint_t i; ngx_shm_zone_t *shm_zone; ngx_list_part_t *part; ngx_slab_pool_t *sp; /* * unlock the accept mutex if the abnormally exited process * held it */ if (ngx_accept_mutex_ptr) { (void) ngx_shmtx_force_unlock(&ngx_accept_mutex, pid); } /* * unlock shared memory mutexes if held by the abnormally exited * process */ part = (ngx_list_part_t *) &ngx_cycle->shared_memory.part; shm_zone = part->elts; for (i = 0; /* void */ ; i++) { if (i >= part->nelts) { if (part->next == NULL) { break; } part = part->next; shm_zone = part->elts; i = 0; } sp = (ngx_slab_pool_t *) shm_zone[i].shm.addr; if (ngx_shmtx_force_unlock(&sp->mutex, pid)) { ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, 0, "shared memory zone \"%V\" was locked by %P", &shm_zone[i].shm.name, pid); } } } void ngx_debug_point(void) { ngx_core_conf_t *ccf; ccf = (ngx_core_conf_t *) ngx_get_conf(ngx_cycle->conf_ctx, ngx_core_module); switch (ccf->debug_points) { case NGX_DEBUG_POINTS_STOP: raise(SIGSTOP); break; case NGX_DEBUG_POINTS_ABORT: ngx_abort(); } } ngx_int_t ngx_os_signal_process(ngx_cycle_t *cycle, char *name, ngx_int_t pid) { ngx_signal_t *sig; for (sig = signals; sig->signo != 0; sig++) { if (ngx_strcmp(name, sig->name) == 0) { if (kill(pid, sig->signo) != -1) { return 0; } ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "kill(%P, %d) failed", pid, sig->signo); } } return 1; }