lighttpd1.4/src/gw_backend.c

#include "first.h"

#include "gw_backend.h"

#include <sys/types.h>
#include "sys-socket.h"
#ifdef HAVE_SYS_UIO_H
#include <sys/uio.h>
#endif
#ifdef HAVE_SYS_WAIT_H
#include <sys/wait.h>
#endif

#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <stdlib.h>
#include <signal.h>
#include <string.h>
#include <unistd.h>

#include "array.h"
#include "buffer.h"
#include "crc32.h"
#include "fdevent.h"
#include "inet_ntop_cache.h"
#include "log.h"




#include "status_counter.h"

static data_integer * gw_status_get_di(server *srv, gw_host *host, gw_proc *proc, const char *tag, size_t len) {
    buffer *b = srv->tmp_buf;
    buffer_copy_string_len(b, CONST_STR_LEN("gw.backend."));
    buffer_append_string_buffer(b, host->id);
    if (proc) {
        buffer_append_string_len(b, CONST_STR_LEN("."));
        buffer_append_int(b, proc->id);
    }
    buffer_append_string_len(b, tag, len);
    return status_counter_get_counter(srv, CONST_BUF_LEN(b));
}

static void gw_proc_tag_inc(server *srv, gw_host *host, gw_proc *proc, const char *tag, size_t len) {
    data_integer *di = gw_status_get_di(srv, host, proc, tag, len);
    ++di->value;
}

static void gw_proc_load_inc(server *srv, gw_host *host, gw_proc *proc) {
    data_integer *di = gw_status_get_di(srv,host,proc,CONST_STR_LEN(".load"));
    di->value = ++proc->load;

    status_counter_inc(srv, CONST_STR_LEN("gw.active-requests"));
}

static void gw_proc_load_dec(server *srv, gw_host *host, gw_proc *proc) {
    data_integer *di = gw_status_get_di(srv,host,proc,CONST_STR_LEN(".load"));
    di->value = --proc->load;

    status_counter_dec(srv, CONST_STR_LEN("gw.active-requests"));
}

static void gw_host_assign(server *srv, gw_host *host) {
    data_integer *di = gw_status_get_di(srv,host,NULL,CONST_STR_LEN(".load"));
    di->value = ++host->load;
}

static void gw_host_reset(server *srv, gw_host *host) {
    data_integer *di = gw_status_get_di(srv,host,NULL,CONST_STR_LEN(".load"));
    di->value = --host->load;
}

static int gw_status_init(server *srv, gw_host *host, gw_proc *proc) {
    gw_status_get_di(srv, host, proc, CONST_STR_LEN(".disabled"))->value = 0;
    gw_status_get_di(srv, host, proc, CONST_STR_LEN(".died"))->value = 0;
    gw_status_get_di(srv, host, proc, CONST_STR_LEN(".overloaded"))->value = 0;
    gw_status_get_di(srv, host, proc, CONST_STR_LEN(".connected"))->value = 0;
    gw_status_get_di(srv, host, proc, CONST_STR_LEN(".load"))->value = 0;

    gw_status_get_di(srv, host, NULL, CONST_STR_LEN(".load"))->value = 0;

    return 0;
}




static void gw_proc_set_state(gw_host *host, gw_proc *proc, int state) {
    if ((int)proc->state == state) return;
    if (proc->state == PROC_STATE_RUNNING) {
        --host->active_procs;
    } else if (state == PROC_STATE_RUNNING) {
        ++host->active_procs;
    }
    proc->state = state;
}


static gw_proc *gw_proc_init(void) {
    gw_proc *f = calloc(1, sizeof(*f));
    force_assert(f);

    f->unixsocket = buffer_init();
    f->connection_name = buffer_init();

    f->prev = NULL;
    f->next = NULL;
    f->state = PROC_STATE_DIED;

    return f;
}

static void gw_proc_free(gw_proc *f) {
    if (!f) return;

    gw_proc_free(f->next);

    buffer_free(f->unixsocket);
    buffer_free(f->connection_name);
    free(f->saddr);

    free(f);
}

static gw_host *gw_host_init(void) {
    gw_host *f = calloc(1, sizeof(*f));
    force_assert(f);

    f->id = buffer_init();
    f->host = buffer_init();
    f->unixsocket = buffer_init();
    f->docroot = buffer_init();
    f->bin_path = buffer_init();
    f->bin_env = array_init();
    f->bin_env_copy = array_init();
    f->strip_request_uri = buffer_init();
    f->xsendfile_docroot = array_init();

    return f;
}

static void gw_host_free(gw_host *h) {
    if (!h) return;
    if (h->refcount) {
        --h->refcount;
        return;
    }

    buffer_free(h->id);
    buffer_free(h->host);
    buffer_free(h->unixsocket);
    buffer_free(h->docroot);
    buffer_free(h->bin_path);
    buffer_free(h->strip_request_uri);
    array_free(h->bin_env);
    array_free(h->bin_env_copy);
    array_free(h->xsendfile_docroot);

    gw_proc_free(h->first);
    gw_proc_free(h->unused_procs);

    for (size_t i = 0; i < h->args.used; ++i) free(h->args.ptr[i]);
    free(h->args.ptr);
    free(h);
}

static gw_exts *gw_extensions_init(void) {
    gw_exts *f = calloc(1, sizeof(*f));
    force_assert(f);
    return f;
}

static void gw_extensions_free(gw_exts *f) {
    if (!f) return;
    for (size_t i = 0; i < f->used; ++i) {
        gw_extension *fe = f->exts[i];
        for (size_t j = 0; j < fe->used; ++j) {
            gw_host_free(fe->hosts[j]);
        }
        buffer_free(fe->key);
        free(fe->hosts);
        free(fe);
    }
    free(f->exts);
    free(f);
}

static int gw_extension_insert(gw_exts *ext, buffer *key, gw_host *fh) {
    gw_extension *fe = NULL;
    for (size_t i = 0; i < ext->used; ++i) {
        if (buffer_is_equal(key, ext->exts[i]->key)) {
            fe = ext->exts[i];
            break;
        }
    }

    if (NULL == fe) {
        fe = calloc(1, sizeof(*fe));
        force_assert(fe);
        fe->key = buffer_init();
        fe->last_used_ndx = -1;
        buffer_copy_buffer(fe->key, key);

        if (ext->size == 0) {
            ext->size = 8;
            ext->exts = malloc(ext->size * sizeof(*(ext->exts)));
            force_assert(ext->exts);
        } else if (ext->used == ext->size) {
            ext->size += 8;
            ext->exts = realloc(ext->exts, ext->size * sizeof(*(ext->exts)));
            force_assert(ext->exts);
        }
        ext->exts[ext->used++] = fe;
        fe->size = 4;
        fe->hosts = malloc(fe->size * sizeof(*(fe->hosts)));
        force_assert(fe->hosts);
    } else if (fe->size == fe->used) {
        fe->size += 4;
        fe->hosts = realloc(fe->hosts, fe->size * sizeof(*(fe->hosts)));
        force_assert(fe->hosts);
    }

    fe->hosts[fe->used++] = fh;
    return 0;
}

static void gw_proc_connect_success(server *srv, gw_host *host, gw_proc *proc, int debug) {
    gw_proc_tag_inc(srv, host, proc, CONST_STR_LEN(".connected"));

    if (debug) {
        log_error_write(srv, __FILE__, __LINE__, "ssdsbsd",
                        "got proc:",
                        "pid:", proc->pid,
                        "socket:", proc->connection_name,
                        "load:", proc->load);
    }
}

static void gw_proc_connect_error(server *srv, gw_host *host, gw_proc *proc, pid_t pid, int errnum, int debug) {
    log_error_write(srv, __FILE__, __LINE__, "sssb",
                    "establishing connection failed:", strerror(errnum),
                    "socket:", proc->connection_name);

    if (!proc->is_local) {
        proc->disabled_until = srv->cur_ts + host->disable_time;
        gw_proc_set_state(host, proc, PROC_STATE_OVERLOADED);
    }
    else if (proc->pid == pid && proc->state == PROC_STATE_RUNNING) {
        /* several requests from lighttpd might reference the same proc
         *
         * Only one of them should mark the proc
         * and all other ones should just take a new one.
         *
         * If a new proc was started with the old struct, this might
         * otherwise lead to marking a perfectly good proc as dead
         */
        log_error_write(srv, __FILE__, __LINE__, "sdssd",
                        "backend error; we'll disable for", host->disable_time,
                        "secs and send the request to another backend instead:",
                        "load:", host->load);
        if (EAGAIN == errnum) {
            /* - EAGAIN: cool down the backend; it is overloaded */
          #ifdef __linux__
            log_error_write(srv, __FILE__, __LINE__, "s",
              "If this happened on Linux: You have run out of local ports. "
              "Check the manual, section Performance how to handle this.");
          #endif
            if (debug) {
                log_error_write(srv, __FILE__, __LINE__, "sbsd",
                  "This means that you have more incoming requests than your "
                  "FastCGI backend can handle in parallel.  It might help to "
                  "spawn more FastCGI backends or PHP children; if not, "
                  "decrease server.max-connections.  The load for this FastCGI "
                  "backend", proc->connection_name, "is", proc->load);
            }
            proc->disabled_until = srv->cur_ts + host->disable_time;
            gw_proc_set_state(host, proc, PROC_STATE_OVERLOADED);
        }
        else {
            /* we got a hard error from the backend like
             * - ECONNREFUSED for tcp-ip sockets
             * - ENOENT for unix-domain-sockets
             */
            gw_proc_set_state(host, proc, PROC_STATE_DIED_WAIT_FOR_PID);
        }
    }

    if (EAGAIN == errnum) {
        gw_proc_tag_inc(srv, host, proc, CONST_STR_LEN(".overloaded"));
    }
    else {
        gw_proc_tag_inc(srv, host, proc, CONST_STR_LEN(".died"));
    }
}

static void gw_proc_release(server *srv, gw_host *host, gw_proc *proc, int debug) {
    gw_proc_load_dec(srv, host, proc);

    if (debug) {
        log_error_write(srv, __FILE__, __LINE__, "ssdsbsd",
                        "released proc:",
                        "pid:", proc->pid,
                        "socket:", proc->connection_name,
                        "load:", proc->load);
    }
}

static void gw_proc_check_enable(server *srv, gw_host *host, gw_proc *proc) {
    if (srv->cur_ts <= proc->disabled_until) return;
    if (proc->state != PROC_STATE_OVERLOADED) return;

    gw_proc_set_state(host, proc, PROC_STATE_RUNNING);

    log_error_write(srv, __FILE__, __LINE__,  "sbbdb",
                    "gw-server re-enabled:", proc->connection_name,
                    host->host, host->port, host->unixsocket);
}

static int gw_proc_waitpid(server *srv, gw_host *host, gw_proc *proc) {
    int rc, status;

    if (!proc->is_local) return 0;
    if (proc->pid <= 0) return 0;

    do {
        rc = waitpid(proc->pid, &status, WNOHANG);
    } while (-1 == rc && errno == EINTR);
    if (0 == rc) return 0; /* child still running */

    /* child terminated */
    if (-1 == rc) {
        /* EINVAL or ECHILD no child processes */
        /* should not happen; someone else has cleaned up for us */
        log_error_write(srv, __FILE__, __LINE__, "sddss",
                        "pid ", proc->pid, proc->state,
                        "not found:", strerror(errno));
    } else if (WIFEXITED(status)) {
        if (proc->state != PROC_STATE_KILLED) {
            log_error_write(srv, __FILE__, __LINE__, "sdb",
                            "child exited:",
                            WEXITSTATUS(status), proc->connection_name);
        }
    } else if (WIFSIGNALED(status)) {
        if (WTERMSIG(status) != SIGTERM && WTERMSIG(status) != SIGINT) {
            log_error_write(srv, __FILE__, __LINE__, "sd",
                            "child signalled:", WTERMSIG(status));
        }
    } else {
        log_error_write(srv, __FILE__, __LINE__, "sd",
                        "child died somehow:", status);
    }

    proc->pid = 0;
    gw_proc_set_state(host, proc, PROC_STATE_DIED);
    return 1;
}

static int gw_proc_sockaddr_init(server *srv, gw_host *host, gw_proc *proc) {
    sock_addr addr;
    socklen_t addrlen;

    if (!buffer_string_is_empty(proc->unixsocket)) {
        if (1 != sock_addr_from_str_hints(srv, &addr, &addrlen,
                                          proc->unixsocket->ptr, AF_UNIX, 0)) {
            errno = EINVAL;
            return -1;
        }
        buffer_copy_string_len(proc->connection_name, CONST_STR_LEN("unix:"));
        buffer_append_string_buffer(proc->connection_name, proc->unixsocket);
    } else {
        if (1 != sock_addr_from_buffer_hints_numeric(srv, &addr, &addrlen,
                                                     host->host, host->family,
                                                     proc->port)) {
            errno = EINVAL;
            return -1;
        }
        buffer_copy_string_len(proc->connection_name, CONST_STR_LEN("tcp:"));
        if (!buffer_string_is_empty(host->host)) {
            buffer_append_string_buffer(proc->connection_name, host->host);
        } else {
            buffer_append_string_len(proc->connection_name,
                                     CONST_STR_LEN("localhost"));
        }
        buffer_append_string_len(proc->connection_name, CONST_STR_LEN(":"));
        buffer_append_int(proc->connection_name, proc->port);
    }

    if (NULL != proc->saddr && proc->saddrlen < addrlen) {
        free(proc->saddr);
        proc->saddr = NULL;
    }
    if (NULL == proc->saddr) {
        proc->saddr = (struct sockaddr *)malloc(addrlen);
        force_assert(proc->saddr);
    }
    proc->saddrlen = addrlen;
    memcpy(proc->saddr, &addr, addrlen);
    return 0;
}

static int env_add(char_array *env, const char *key, size_t key_len, const char *val, size_t val_len) {
    char *dst;

    if (!key || !val) return -1;

    dst = malloc(key_len + val_len + 3);
    force_assert(dst);
    memcpy(dst, key, key_len);
    dst[key_len] = '=';
    memcpy(dst + key_len + 1, val, val_len + 1); /* add the \0 from the value */

    for (size_t i = 0; i < env->used; ++i) {
        if (0 == strncmp(dst, env->ptr[i], key_len + 1)) {
            free(env->ptr[i]);
            env->ptr[i] = dst;
            return 0;
        }
    }

    if (env->size == 0) {
        env->size = 16;
        env->ptr = malloc(env->size * sizeof(*env->ptr));
        force_assert(env->ptr);
    } else if (env->size == env->used + 1) {
        env->size += 16;
        env->ptr = realloc(env->ptr, env->size * sizeof(*env->ptr));
        force_assert(env->ptr);
    }

    env->ptr[env->used++] = dst;

    return 0;
}

static int gw_spawn_connection(server *srv, gw_host *host, gw_proc *proc, int debug) {
    int gw_fd;
    int status;
    struct timeval tv = { 0, 10 * 1000 };

    if (debug) {
        log_error_write(srv, __FILE__, __LINE__, "sdb",
                        "new proc, socket:", proc->port, proc->unixsocket);
    }

    gw_fd = fdevent_socket_cloexec(proc->saddr->sa_family, SOCK_STREAM, 0);
    if (-1 == gw_fd) {
        log_error_write(srv, __FILE__, __LINE__, "ss",
                        "failed:", strerror(errno));
        return -1;
    }

    do {
        status = connect(gw_fd, proc->saddr, proc->saddrlen);
    } while (-1 == status && errno == EINTR);

    if (-1 == status && errno != ENOENT
        && !buffer_string_is_empty(proc->unixsocket)) {
        log_error_write(srv, __FILE__, __LINE__, "sbss",
                        "unlink", proc->unixsocket,
                        "after connect failed:", strerror(errno));
        unlink(proc->unixsocket->ptr);
    }

    close(gw_fd);

    if (-1 == status) {
        /* server is not up, spawn it  */
        char_array env;
        size_t i;
        int val;
        int dfd = -1;

        /* reopen socket */
        gw_fd = fdevent_socket_cloexec(proc->saddr->sa_family, SOCK_STREAM, 0);
        if (-1 == gw_fd) {
            log_error_write(srv, __FILE__, __LINE__, "ss",
                            "socket failed:", strerror(errno));
            return -1;
        }

        val = 1;
        if (setsockopt(gw_fd,SOL_SOCKET,SO_REUSEADDR,&val,sizeof(val)) < 0) {
            log_error_write(srv, __FILE__, __LINE__, "ss",
                            "socketsockopt failed:", strerror(errno));
            close(gw_fd);
            return -1;
        }

        /* create socket */
        if (-1 == bind(gw_fd, proc->saddr, proc->saddrlen)) {
            log_error_write(srv, __FILE__, __LINE__, "sbs",
                            "bind failed for:",
                            proc->connection_name,
                            strerror(errno));
            close(gw_fd);
            return -1;
        }

        if (-1 == listen(gw_fd, host->listen_backlog)) {
            log_error_write(srv, __FILE__, __LINE__, "ss",
                            "listen failed:", strerror(errno));
            close(gw_fd);
            return -1;
        }

        {
            /* create environment */
            env.ptr = NULL;
            env.size = 0;
            env.used = 0;

            /* build clean environment */
            if (host->bin_env_copy->used) {
                for (i = 0; i < host->bin_env_copy->used; ++i) {
                    data_string *ds=(data_string *)host->bin_env_copy->data[i];
                    char *ge;

                    if (NULL != (ge = getenv(ds->value->ptr))) {
                        env_add(&env, CONST_BUF_LEN(ds->value), ge, strlen(ge));
                    }
                }
            } else {
                char ** const e = environ;
                for (i = 0; e[i]; ++i) {
                    char *eq;

                    if (NULL != (eq = strchr(e[i], '='))) {
                        env_add(&env, e[i], eq - e[i], eq+1, strlen(eq+1));
                    }
                }
            }

            /* create environment */
            for (i = 0; i < host->bin_env->used; ++i) {
                data_string *ds = (data_string *)host->bin_env->data[i];

                env_add(&env, CONST_BUF_LEN(ds->key), CONST_BUF_LEN(ds->value));
            }

            for (i = 0; i < env.used; ++i) {
                /* search for PHP_FCGI_CHILDREN */
                if (0 == strncmp(env.ptr[i], "PHP_FCGI_CHILDREN=",
                                      sizeof("PHP_FCGI_CHILDREN=")-1)) {
                    break;
                }
            }

            /* not found, add a default */
            if (i == env.used) {
                env_add(&env, CONST_STR_LEN("PHP_FCGI_CHILDREN"),
                              CONST_STR_LEN("1"));
            }

            env.ptr[env.used] = NULL;
        }

        dfd = fdevent_open_dirname(host->args.ptr[0]);
        if (-1 == dfd) {
            log_error_write(srv, __FILE__, __LINE__, "sss",
                            "open dirname failed:", strerror(errno),
                            host->args.ptr[0]);
        }

        /*(FCGI_LISTENSOCK_FILENO == STDIN_FILENO == 0)*/
        proc->pid = (dfd >= 0)
          ? fdevent_fork_execve(host->args.ptr[0], host->args.ptr,
                                env.ptr, gw_fd, -1, -1, dfd)
          : -1;

        for (i = 0; i < env.used; ++i) free(env.ptr[i]);
        free(env.ptr);
        if (-1 != dfd) close(dfd);
        close(gw_fd);

        if (-1 == proc->pid) {
            log_error_write(srv, __FILE__, __LINE__, "sb",
                            "gw-backend failed to start:", host->bin_path);
            return -1;
        }

        /* register process */
        proc->last_used = srv->cur_ts;
        proc->is_local = 1;

        /* wait */
        select(0, NULL, NULL, NULL, &tv);

        if (0 != gw_proc_waitpid(srv, host, proc)) {
            log_error_write(srv, __FILE__, __LINE__, "sb",
                            "gw-backend failed to start:", host->bin_path);
            log_error_write(srv, __FILE__, __LINE__, "s",
              "If you're trying to run your app as a FastCGI backend, make "
              "sure you're using the FastCGI-enabled version.  If this is PHP "
              "on Gentoo, add 'fastcgi' to the USE flags.  If this is PHP, try "
              "removing the bytecode caches for now and try again.");
            return -1;
        }
    } else {
        proc->is_local = 0;
        proc->pid = 0;

        if (debug) {
            log_error_write(srv, __FILE__, __LINE__, "sb",
                            "(debug) socket is already used; won't spawn:",
                            proc->connection_name);
        }
    }

    gw_proc_set_state(host, proc, PROC_STATE_RUNNING);
    return 0;
}

static void gw_proc_spawn(server *srv, gw_host *host, int debug) {
    gw_proc *proc;
    for (proc=host->unused_procs; proc && proc->pid != 0; proc=proc->next);
    if (proc) {
        if (proc == host->unused_procs)
            host->unused_procs = proc->next;
        else
            proc->prev->next = proc->next;

        if (proc->next) {
            proc->next->prev = proc->prev;
            proc->next = NULL;
        }

        proc->prev = NULL;
    } else {
        proc = gw_proc_init();
        proc->id = host->max_id++;
    }

    ++host->num_procs;

    if (buffer_string_is_empty(host->unixsocket)) {
        proc->port = host->port + proc->id;
    } else {
        buffer_copy_buffer(proc->unixsocket, host->unixsocket);
        buffer_append_string_len(proc->unixsocket, CONST_STR_LEN("-"));
        buffer_append_int(proc->unixsocket, proc->id);
    }

    if (0 != gw_proc_sockaddr_init(srv, host, proc)) {
        /*(should not happen if host->host validated at startup,
         * and translated from name to IP address at startup)*/
        log_error_write(srv, __FILE__, __LINE__, "s",
                        "ERROR: spawning backend failed.");
        --host->num_procs;
        if (proc->id == host->max_id-1) --host->max_id;
        gw_proc_free(proc);
    } else if (gw_spawn_connection(srv, host, proc, debug)) {
        log_error_write(srv, __FILE__, __LINE__, "s",
                        "ERROR: spawning backend failed.");
        proc->next = host->unused_procs;
        if (host->unused_procs)
            host->unused_procs->prev = proc;
        host->unused_procs = proc;
    } else {
        proc->next = host->first;
        if (host->first)
            host->first->prev = proc;
        host->first = proc;
    }
}

static void gw_proc_kill(server *srv, gw_host *host, gw_proc *proc) {
    UNUSED(srv);
    if (proc->next) proc->next->prev = proc->prev;
    if (proc->prev) proc->prev->next = proc->next;

    if (proc->prev == NULL) host->first = proc->next;

    proc->prev = NULL;
    proc->next = host->unused_procs;

    if (host->unused_procs)
        host->unused_procs->prev = proc;
    host->unused_procs = proc;

    kill(proc->pid, SIGTERM);

    gw_proc_set_state(host, proc, PROC_STATE_KILLED);

    --host->num_procs;
}

static gw_host * unixsocket_is_dup(gw_plugin_data *p, size_t used, buffer *unixsocket) {
    for (size_t i = 0; i < used; ++i) {
        gw_exts *exts = p->config_storage[i]->exts;
        if (NULL == exts) continue;
        for (size_t j = 0; j < exts->used; ++j) {
            gw_extension *ex = exts->exts[j];
            for (size_t n = 0; n < ex->used; ++n) {
                gw_host *host = ex->hosts[n];
                if (!buffer_string_is_empty(host->unixsocket)
                    && buffer_is_equal(host->unixsocket, unixsocket)
                    && !buffer_string_is_empty(host->bin_path))
                    return host;
            }
        }
    }

    return NULL;
}

static int parse_binpath(char_array *env, buffer *b) {
    char *start = b->ptr;
    char c;
    /* search for spaces */
    for (size_t i = 0; i < buffer_string_length(b); ++i) {
        switch(b->ptr[i]) {
        case ' ':
        case '\t':
            /* a WS, stop here and copy the argument */

            if (env->size == 0) {
                env->size = 16;
                env->ptr = malloc(env->size * sizeof(*env->ptr));
            } else if (env->size == env->used) {
                env->size += 16;
                env->ptr = realloc(env->ptr, env->size * sizeof(*env->ptr));
            }

            c = b->ptr[i];
            b->ptr[i] = '\0';
            env->ptr[env->used++] = strdup(start);
            b->ptr[i] = c;

            start = b->ptr + i + 1;
            break;
        default:
            break;
        }
    }

    if (env->size == 0) {
        env->size = 16;
        env->ptr = malloc(env->size * sizeof(*env->ptr));
    } else if (env->size == env->used) { /*need one extra for terminating NULL*/
        env->size += 16;
        env->ptr = realloc(env->ptr, env->size * sizeof(*env->ptr));
    }

    /* the rest */
    env->ptr[env->used++] = strdup(start);

    if (env->size == 0) {
        env->size = 16;
        env->ptr = malloc(env->size * sizeof(*env->ptr));
    } else if (env->size == env->used) { /*need one extra for terminating NULL*/
        env->size += 16;
        env->ptr = realloc(env->ptr, env->size * sizeof(*env->ptr));
    }

    /* terminate */
    env->ptr[env->used++] = NULL;

    return 0;
}

enum {
  GW_BALANCE_LEAST_CONNECTION,
  GW_BALANCE_RR,
  GW_BALANCE_HASH,
  GW_BALANCE_STICKY
};

static gw_host * gw_host_get(server *srv, connection *con, gw_extension *extension, int balance, int debug) {
    gw_host *host;
    unsigned long last_max = ULONG_MAX;
    int max_usage = INT_MAX;
    int ndx = -1;
    size_t k;

    if (extension->used <= 1) {
        if (1 == extension->used && extension->hosts[0]->active_procs > 0) {
            ndx = 0;
        }
    } else switch(balance) {
    case GW_BALANCE_HASH:
        /* hash balancing */

        if (debug) {
            log_error_write(srv, __FILE__, __LINE__,  "sd",
                            "proxy - used hash balancing, hosts:",
                            extension->used);
        }

        for (k = 0, ndx = -1, last_max = ULONG_MAX; k < extension->used; ++k) {
            unsigned long cur_max;
            host = extension->hosts[k];
            if (0 == host->active_procs) continue;

            cur_max = generate_crc32c(CONST_BUF_LEN(con->uri.path))
                    + generate_crc32c(CONST_BUF_LEN(host->host)) /* cachable */
                    + generate_crc32c(CONST_BUF_LEN(con->uri.authority));

            if (debug) {
                log_error_write(srv, __FILE__, __LINE__,  "sbbbd",
                                "proxy - election:", con->uri.path,
                                host->host, con->uri.authority, cur_max);
            }

            if (last_max < cur_max || last_max == ULONG_MAX) {
                last_max = cur_max;
                ndx = k;
            }
        }

        break;
    case GW_BALANCE_LEAST_CONNECTION:
        /* fair balancing */
        if (debug) {
            log_error_write(srv, __FILE__, __LINE__,  "s",
                            "proxy - used least connection");
        }

        for (k = 0, ndx = -1, max_usage = INT_MAX; k < extension->used; ++k) {
            host = extension->hosts[k];
            if (0 == host->active_procs) continue;

            if (host->load < max_usage) {
                max_usage = host->load;
                ndx = k;
            }
        }

        break;
    case GW_BALANCE_RR:
        /* round robin */
        if (debug) {
            log_error_write(srv, __FILE__, __LINE__,  "s",
                            "proxy - used round-robin balancing");
        }

        /* just to be sure */
        force_assert(extension->used < INT_MAX);

        host = extension->hosts[0];

        /* Use last_used_ndx from first host in list */
        k = extension->last_used_ndx;
        ndx = k + 1; /* use next host after the last one */
        if (ndx < 0) ndx = 0;

        /* Search first active host after last_used_ndx */
        while (ndx < (int) extension->used
               && 0 == (host = extension->hosts[ndx])->active_procs) ++ndx;

        if (ndx >= (int) extension->used) {
            /* didn't find a higher id, wrap to the start */
            for (ndx = 0; ndx <= (int) k; ++ndx) {
                host = extension->hosts[ndx];
                if (0 != host->active_procs) break;
            }

            /* No active host found */
            if (0 == host->active_procs) ndx = -1;
        }

        /* Save new index for next round */
        extension->last_used_ndx = ndx;

        break;
    case GW_BALANCE_STICKY:
        /* source sticky balancing */

        if (debug) {
            log_error_write(srv, __FILE__, __LINE__,  "sd",
                            "proxy - used sticky balancing, hosts:",
                            extension->used);
        }

        for (k = 0, ndx = -1, last_max = ULONG_MAX; k < extension->used; ++k) {
            unsigned long cur_max;
            host = extension->hosts[k];

            if (0 == host->active_procs) continue;

            cur_max = generate_crc32c(CONST_BUF_LEN(con->dst_addr_buf))
                    + generate_crc32c(CONST_BUF_LEN(host->host))
                    + host->port;

            if (debug) {
                log_error_write(srv, __FILE__, __LINE__,  "sbbdd",
                                "proxy - election:", con->dst_addr_buf,
                                host->host, host->port, cur_max);
            }

            if (last_max < cur_max || last_max == ULONG_MAX) {
                last_max = cur_max;
                ndx = k;
            }
        }

        break;
    default:
        break;
    }

    if (-1 != ndx) {
        /* found a server */
        host = extension->hosts[ndx];

        if (debug) {
            log_error_write(srv, __FILE__, __LINE__,  "sbd",
                            "gw - found a host", host->host, host->port);
        }

        return host;
    }

    /* all hosts are down */
    /* sorry, we don't have a server alive for this ext */
    con->http_status = 503; /* Service Unavailable */
    con->mode = DIRECT;

    /* only send the 'no handler' once */
    if (!extension->note_is_sent) {
        extension->note_is_sent = 1;
        log_error_write(srv, __FILE__, __LINE__, "sBSbsbs",
                        "all handlers for", con->uri.path, "?",
                        con->uri.query, "on", extension->key, "are down.");
    }

    return NULL;
}

static int gw_establish_connection(server *srv, gw_host *host, gw_proc *proc, pid_t pid, int gw_fd, int debug) {
    if (-1 == connect(gw_fd, proc->saddr, proc->saddrlen)) {
        if (errno == EINPROGRESS ||
            errno == EALREADY ||
            errno == EINTR) {
            if (debug > 2) {
                log_error_write(srv, __FILE__, __LINE__, "sb",
                                "connect delayed; will continue later:",
                                proc->connection_name);
            }

            return 1;
        } else {
            gw_proc_connect_error(srv, host, proc, pid, errno, debug);
            return -1;
        }
    }

    if (debug > 1) {
        log_error_write(srv, __FILE__, __LINE__, "sd",
                        "connect succeeded: ", gw_fd);
    }

    return 0;
}

static void gw_restart_dead_procs(server *srv, gw_host *host, int debug) {
    for (gw_proc *proc = host->first; proc; proc = proc->next) {
        if (debug > 2) {
            log_error_write(srv, __FILE__, __LINE__,  "sbdddd",
                            "proc:", proc->connection_name, proc->state,
                            proc->is_local, proc->load, proc->pid);
        }

        /*
         * if the remote side is overloaded, we check back after <n> seconds
         *
         */
        switch (proc->state) {
        case PROC_STATE_KILLED:
            /* should never happen as long as adaptive spawing is disabled */
            force_assert(0);

            break;
        case PROC_STATE_RUNNING:
            break;
        case PROC_STATE_OVERLOADED:
        case PROC_STATE_DIED_WAIT_FOR_PID:
            if (0 == gw_proc_waitpid(srv, host, proc)) {
                gw_proc_check_enable(srv, host, proc);
            }

            if (proc->state != PROC_STATE_DIED) break;
            /* fall through *//*(we have a dead proc now)*/

        case PROC_STATE_DIED:
            /* local procs get restarted by us,
             * remote ones hopefully by the admin */

            if (!buffer_string_is_empty(host->bin_path)) {
                /* we still have connections bound to this proc,
                 * let them terminate first */
                if (proc->load != 0) break;

                /* restart the child */

                if (debug) {
                    log_error_write(srv, __FILE__, __LINE__, "ssbsdsd",
                                    "--- gw spawning",
                                    "\n\tsocket", proc->connection_name,
                                    "\n\tcurrent:", 1, "/", host->max_procs);
                }

                if (gw_spawn_connection(srv, host, proc, debug)) {
                    log_error_write(srv, __FILE__, __LINE__, "s",
                                    "ERROR: spawning gw failed.");
                }
            } else {
                gw_proc_check_enable(srv, host, proc);
            }
            break;
        }
    }
}




#include "base.h"
#include "connections.h"
#include "joblist.h"
#include "keyvalue.h"
#include "plugin.h"
#include "response.h"


/* ok, we need a prototype */
static handler_t gw_handle_fdevent(server *srv, void *ctx, int revents);


static gw_handler_ctx * handler_ctx_init(size_t sz) {
    gw_handler_ctx *hctx = calloc(1, 0 == sz ? sizeof(*hctx) : sz);
    force_assert(hctx);

    hctx->fde_ndx = -1;

    /*hctx->response = buffer_init();*//*(allocated when needed)*/

    hctx->request_id = 0;
    hctx->gw_mode = GW_RESPONDER;
    hctx->state = GW_STATE_INIT;
    hctx->proc = NULL;

    hctx->fd = -1;

    hctx->reconnects = 0;
    hctx->send_content_body = 1;

    /*hctx->rb = chunkqueue_init();*//*(allocated when needed)*/
    hctx->wb = chunkqueue_init();
    hctx->wb_reqlen = 0;

    return hctx;
}