leader.c

apache的软件linux版本

/* Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 "apr.h"
#include "apr_portable.h"
#include "apr_strings.h"
#include "apr_file_io.h"
#include "apr_thread_proc.h"
#include "apr_signal.h"
#include "apr_thread_cond.h"
#include "apr_thread_mutex.h"
#include "apr_proc_mutex.h"
#define APR_WANT_STRFUNC
#include "apr_want.h"

#if APR_HAVE_UNISTD_H
#include <unistd.h>
#endif
#if APR_HAVE_SYS_SOCKET_H
#include <sys/socket.h>
#endif
#if APR_HAVE_SYS_WAIT_H
#include <sys/wait.h> 
#endif
#ifdef HAVE_SYS_PROCESSOR_H
#include <sys/processor.h> /* for bindprocessor() */
#endif

#if !APR_HAS_THREADS
#error The Leader/Follower MPM requires APR threads, but they are unavailable.
#endif

#define CORE_PRIVATE 
 
#include "ap_config.h"
#include "httpd.h" 
#include "http_main.h" 
#include "http_log.h" 
#include "http_config.h"        /* for read_config */ 
#include "http_core.h"          /* for get_remote_host */ 
#include "http_connection.h"
#include "ap_mpm.h"
#include "mpm_common.h"
#include "ap_listen.h"
#include "scoreboard.h" 
#include "mpm_default.h"
#include "apr_poll.h"

#include <signal.h>
#include <limits.h>             /* for INT_MAX */

#include "apr_atomic.h"

/* Limit on the total --- clients will be locked out if more servers than
 * this are needed.  It is intended solely to keep the server from crashing
 * when things get out of hand.
 *
 * We keep a hard maximum number of servers, for two reasons --- first off,
 * in case something goes seriously wrong, we want to stop the fork bomb
 * short of actually crashing the machine we're running on by filling some
 * kernel table.  Secondly, it keeps the size of the scoreboard file small
 * enough that we can read the whole thing without worrying too much about
 * the overhead.
 */
#ifndef DEFAULT_SERVER_LIMIT
#define DEFAULT_SERVER_LIMIT 16
#endif

/* Admin can't tune ServerLimit beyond MAX_SERVER_LIMIT.  We want
 * some sort of compile-time limit to help catch typos.
 */
#ifndef MAX_SERVER_LIMIT
#define MAX_SERVER_LIMIT 20000
#endif

/* Limit on the threads per process.  Clients will be locked out if more than
 * this  * server_limit are needed.
 *
 * We keep this for one reason it keeps the size of the scoreboard file small
 * enough that we can read the whole thing without worrying too much about
 * the overhead.
 */
#ifndef DEFAULT_THREAD_LIMIT
#define DEFAULT_THREAD_LIMIT 64 
#endif

/* Admin can't tune ThreadLimit beyond MAX_THREAD_LIMIT.  We want
 * some sort of compile-time limit to help catch typos.
 */
#ifndef MAX_THREAD_LIMIT
#define MAX_THREAD_LIMIT 20000
#endif

/*
 * Actual definitions of config globals
 */

int ap_threads_per_child = 0;         /* Worker threads per child */
static int ap_daemons_to_start = 0;
static int min_spare_threads = 0;
static int max_spare_threads = 0;
static int ap_daemons_limit = 0;
static int server_limit = DEFAULT_SERVER_LIMIT;
static int first_server_limit;
static int thread_limit = DEFAULT_THREAD_LIMIT;
static int first_thread_limit;
static int changed_limit_at_restart;
static int dying = 0;
static int workers_may_exit = 0;
static int start_thread_may_exit = 0;
static int requests_this_child;
static int num_listensocks = 0;
static int resource_shortage = 0;
static int mpm_state = AP_MPMQ_STARTING;

typedef struct worker_wakeup_info worker_wakeup_info;

/* The structure used to pass unique initialization info to each thread */
typedef struct {
    int pid;
    int tid;
    int sd;
} proc_info;


/* Structure used to pass information to the thread responsible for 
 * creating the rest of the threads.
 */
typedef struct {
    apr_thread_t **threads;
    int child_num_arg;
    apr_threadattr_t *threadattr;
} thread_starter;

#define ID_FROM_CHILD_THREAD(c, t)    ((c * thread_limit) + t)

/*
 * The max child slot ever assigned, preserved across restarts.  Necessary
 * to deal with MaxClients changes across AP_SIG_GRACEFUL restarts.  We 
 * use this value to optimize routines that have to scan the entire 
 * scoreboard.
 */
int ap_max_daemons_limit = -1;

static ap_pod_t *pod;

/* *Non*-shared http_main globals... */

server_rec *ap_server_conf;

/* This MPM respects a couple of runtime flags that can aid in debugging.
 *  Setting the -DNO_DETACH flag will prevent the root process from
 *  detaching from its controlling terminal. Additionally, setting
 * the -DONE_PROCESS flag (which implies -DNO_DETACH) will get you the
 * child_main loop running in the process which originally started up.
 * This gives you a pretty nice debugging environment.  (You'll get a SIGHUP
 * early in standalone_main; just continue through.  This is the server
 * trying to kill off any child processes which it might have lying
 * around --- Apache doesn't keep track of their pids, it just sends
 * SIGHUP to the process group, ignoring it in the root process.
 * Continue through and you'll be fine.).
 */

static int one_process = 0;

#ifdef DEBUG_SIGSTOP
int raise_sigstop_flags;
#endif

static apr_pool_t *pconf;                 /* Pool for config stuff */
static apr_pool_t *pchild;                /* Pool for httpd child stuff */

static pid_t ap_my_pid; /* Linux getpid() doesn't work except in main 
                           thread. Use this instead */
static pid_t parent_pid;

/* Locks for accept serialization */
static apr_proc_mutex_t *accept_mutex;

#ifdef SINGLE_LISTEN_UNSERIALIZED_ACCEPT
#define SAFE_ACCEPT(stmt) (ap_listeners->next ? (stmt) : APR_SUCCESS)
#else
#define SAFE_ACCEPT(stmt) (stmt)
#endif


/* Structure used to wake up an idle worker thread
 */
struct worker_wakeup_info {
    apr_uint32_t next; /* index into worker_wakeups array,
                        * used to build a linked list
                        */
    apr_thread_cond_t *cond;
    apr_thread_mutex_t *mutex;
};

static worker_wakeup_info *worker_wakeup_create(apr_pool_t *pool)
{
    apr_status_t rv;
    worker_wakeup_info *wakeup;

    wakeup = (worker_wakeup_info *)apr_palloc(pool, sizeof(*wakeup));
    if ((rv = apr_thread_cond_create(&wakeup->cond, pool)) != APR_SUCCESS) {
        return NULL;
    }
    if ((rv = apr_thread_mutex_create(&wakeup->mutex, APR_THREAD_MUTEX_DEFAULT,
                                      pool)) != APR_SUCCESS) {
        return NULL;
    }
    /* The wakeup's mutex will be unlocked automatically when
     * the worker blocks on the condition variable
     */
    apr_thread_mutex_lock(wakeup->mutex);
    return wakeup;
}


/* Structure used to hold a stack of idle worker threads 
 */
typedef struct {
    /* 'state' consists of several fields concatenated into a
     * single 32-bit int for use with the apr_atomic_cas() API:
     *   state & STACK_FIRST  is the thread ID of the first thread
     *                        in a linked list of idle threads
     *   state & STACK_TERMINATED  indicates whether the proc is shutting down
     *   state & STACK_NO_LISTENER indicates whether the process has
     *                             no current listener thread
     */
    apr_uint32_t state;
} worker_stack;

#define STACK_FIRST  0xffff
#define STACK_LIST_END  0xffff
#define STACK_TERMINATED 0x10000
#define STACK_NO_LISTENER 0x20000

static worker_wakeup_info **worker_wakeups = NULL;

static worker_stack* worker_stack_create(apr_pool_t *pool, apr_size_t max)
{
    worker_stack *stack = (worker_stack *)apr_palloc(pool, sizeof(*stack));
    stack->state = STACK_NO_LISTENER | STACK_LIST_END;
    return stack;
}

static apr_status_t worker_stack_wait(worker_stack *stack,
                                      apr_uint32_t worker_id)
{
    worker_wakeup_info *wakeup = worker_wakeups[worker_id];

    while (1) {
        apr_uint32_t state = stack->state;
        if (state & (STACK_TERMINATED | STACK_NO_LISTENER)) {
            if (state & STACK_TERMINATED) {
                return APR_EINVAL;
            }
            if (apr_atomic_cas(&(stack->state), STACK_LIST_END, state) !=
                state) {
                continue;
            }
            else {
                return APR_SUCCESS;
            }
        }
        wakeup->next = state;
        if (apr_atomic_cas(&(stack->state), worker_id, state) != state) {
            continue;
        }
        else {
            return apr_thread_cond_wait(wakeup->cond, wakeup->mutex);
        }
    }    
}

static apr_status_t worker_stack_awaken_next(worker_stack *stack)
{

    while (1) {
        apr_uint32_t state = stack->state;
        apr_uint32_t first = state & STACK_FIRST;
        if (first == STACK_LIST_END) {
            if (apr_atomic_cas(&(stack->state), state | STACK_NO_LISTENER,
                               state) != state) {
                continue;
            }
            else {
                return APR_SUCCESS;
            }
        }
        else {
            worker_wakeup_info *wakeup = worker_wakeups[first];
            if (apr_atomic_cas(&(stack->state), (state ^ first) | wakeup->next,
                               state) != state) {
                continue;
            }
            else {
                /* Acquire and release the idle worker's mutex to ensure
                 * that it's actually waiting on its condition variable
                 */
                apr_status_t rv;
                if ((rv = apr_thread_mutex_lock(wakeup->mutex)) !=
                    APR_SUCCESS) {
                    return rv;
                }
                if ((rv = apr_thread_mutex_unlock(wakeup->mutex)) !=
                    APR_SUCCESS) {
                    return rv;
                }
                return apr_thread_cond_signal(wakeup->cond);
            }
        }
    }
}

static apr_status_t worker_stack_term(worker_stack *stack)
{
    int i;
    apr_status_t rv;

    while (1) {
        apr_uint32_t state = stack->state;
        if (apr_atomic_cas(&(stack->state), state | STACK_TERMINATED,
                           state) == state) {
            break;
        }
    }
    for (i = 0; i < ap_threads_per_child; i++) {
        if ((rv = worker_stack_awaken_next(stack)) != APR_SUCCESS) {
            return rv;
        }
    }
    return APR_SUCCESS;
}

static worker_stack *idle_worker_stack;

#define ST_INIT              0
#define ST_GRACEFUL          1
#define ST_UNGRACEFUL        2

static int terminate_mode = ST_INIT;

static void signal_threads(int mode)
{
    if (terminate_mode == mode) {
        return;
    }
    terminate_mode = mode;
    mpm_state = AP_MPMQ_STOPPING;
    workers_may_exit = 1;

    worker_stack_term(idle_worker_stack);
}

AP_DECLARE(apr_status_t) ap_mpm_query(int query_code, int *result)
{
    switch(query_code){
        case AP_MPMQ_MAX_DAEMON_USED:
            *result = ap_max_daemons_limit;
            return APR_SUCCESS;
        case AP_MPMQ_IS_THREADED:
            *result = AP_MPMQ_STATIC;
            return APR_SUCCESS;