Commit 08daed2b authored by Brian Pane's avatar Brian Pane
Browse files

Added asynchronous keep-alive support to the leader MPM... 

the new io_multiplexer object is intended to provide a
foundation for async processing of the other connection
states.


git-svn-id: https://svn.apache.org/repos/asf/httpd/httpd/branches/async-dev@280222 13f79535-47bb-0310-9956-ffa450edef68
parent 1e90fd1a

server/mpm/experimental/leader/Makefile.in

+1 −1
Original line number Diff line number Diff line 


LTLIBRARY_NAME    = libleader.la

LTLIBRARY_SOURCES = leader.c

LTLIBRARY_SOURCES = leader.c io_multiplexer.c



include $(top_srcdir)/build/ltlib.mk

server/mpm/experimental/leader/README

+18 −9
Original line number Diff line number Diff line 
Leader MPM:

This is an experimental variant of the standard worker MPM.

It uses a Leader/Followers design pattern to coordinate work among threads:



This is an experimental MPM that uses the Leader/Followers design

pattern to coordinate work among threads:

http://deuce.doc.wustl.edu/doc/pspdfs/lf.pdf



As of httpd-2.3, the Leader MPM also incorporates a variant of the

Event MPM's asynchronous socket I/O for keepalive connections.  The

management of pollsets and timeouts is encapsulated within the

io_multiplexer functions, with the aim of eventually supporting

asynchronous write completion.



To use the leader MPM, add "--with-mpm=leader" to the configure

script's arguments when building the httpd.

  

This MPM depends on APR's atomic compare-and-swap operations for

thread synchronization.  If you are compiling for an x86 target

and you don't need to support 386s, or you're compiling for a

SPARC and you don't need to run on pre-UltraSPARC chips, add

"--enable-nonportable-atomics=yes" to the configure script's

arguments.  This will cause APR to implement atomic operations

using efficient opcodes not available in older CPUs.
 
IMPORTANT NOTES:



* At the moment, with the async code under active development, the

  Leader MPM is suitable for R&D use, but not for production use.



* Like the Event MPM, the Leader MPM now requires a threadsafe

  apr_pollset implementation, such as epoll on Linux or kqueue

  on BSD.

server/mpm/experimental/leader/io_multiplexer.c

0 → 100644
+254 −0
Original line number Diff line number Diff line 
/* Copyright 2005 The Apache Software Foundation or its licensors, as

 * applicable.

 *

 * Licensed 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_poll.h"

#include "apr_ring.h"

#include "apr_thread_cond.h"

#include "apr_thread_mutex.h"



#include "io_multiplexer.h"



extern server_rec *ap_server_conf;



APR_RING_HEAD(timeout_ring_header_t, conn_state_t);



struct io_multiplexer {

    int stopped;

    apr_thread_mutex_t *lock;

    apr_thread_mutex_t *pollset_lock;

    apr_pollset_t *pollset;

    apr_int32_t num_pending_events;

    const apr_pollfd_t *next_pending_event;

    struct timeout_ring_header_t pending_timeouts;

    struct timeout_ring_header_t expired_timeouts;

    volatile int poll_sequence_num;

};



static apr_status_t io_multiplexer_remove_internal(io_multiplexer *iom,

                                                   multiplexable *m);



apr_status_t io_multiplexer_create(io_multiplexer **iom, apr_pool_t *p,

                                   apr_uint32_t max_descriptors)

{

    apr_status_t rv;

    *iom = (io_multiplexer *)apr_palloc(p, sizeof(**iom));

    rv = apr_thread_mutex_create(&((*iom)->lock), APR_THREAD_MUTEX_DEFAULT, p);

    if (rv != APR_SUCCESS) {

        return rv;

    }

    rv = apr_thread_mutex_create(&((*iom)->pollset_lock),

                                 APR_THREAD_MUTEX_DEFAULT, p);

    if (rv != APR_SUCCESS) {

        return rv;

    }

    rv = apr_pollset_create(&((*iom)->pollset), max_descriptors, p,

                            APR_POLLSET_THREADSAFE);

    if (rv != APR_SUCCESS) {

        return rv;

    }

    (*iom)->stopped = 0;

    (*iom)->num_pending_events = 0;

    (*iom)->next_pending_event = NULL;

    APR_RING_INIT(&((*iom)->pending_timeouts), conn_state_t, timeout_list);

    APR_RING_INIT(&((*iom)->expired_timeouts), conn_state_t, timeout_list);

    (*iom)->poll_sequence_num = 0;

    

    return APR_SUCCESS;

}



#define DEFAULT_POLL_TIMEOUT 1000000



apr_status_t io_multiplexer_get_event(io_multiplexer *iom,

                                      apr_pollfd_t *event)

{

    apr_status_t rv;

    rv = apr_thread_mutex_lock(iom->pollset_lock);

    if (rv != APR_SUCCESS) {

        return rv;

    }

    rv = apr_thread_mutex_lock(iom->lock);

    if (rv != APR_SUCCESS) {

        apr_thread_mutex_unlock(iom->pollset_lock);

        return rv;

    }

    if (iom->stopped) {

        apr_thread_mutex_unlock(iom->lock);

        apr_thread_mutex_unlock(iom->pollset_lock);

        return APR_EINVAL;

    }

    for (;;) {

        /* Invariant: at the start of each iteration of this loop, the

         * active thread holds iom->lock.

         */

         if (!APR_RING_EMPTY(&(iom->expired_timeouts), conn_state_t, timeout_list)) {

             /* There are some timeout notifications remaining

              * from the last poll.  Return the next one.

              */

             conn_state_t *cs = APR_RING_FIRST(&(iom->expired_timeouts));

             APR_RING_REMOVE(cs, timeout_list);

             *event = cs->pfd;

             event->rtnevents |= IOM_POLL_TIMEOUT;

             apr_thread_mutex_unlock(iom->lock);

             return apr_thread_mutex_unlock(iom->pollset_lock);

         }

         else if (iom->num_pending_events > 0) {

            /* There are some events remaining from the last

             * poll.  Return the next one.

             */

            *event = *(iom->next_pending_event++);

            apr_pollset_remove(iom->pollset, event);

            iom->num_pending_events--;

            apr_thread_mutex_unlock(iom->lock);

            return apr_thread_mutex_unlock(iom->pollset_lock);

        }

        else {

            /* No unprocessed events remain from the previous poll,

             * so initiate a new poll.

             */

            apr_int32_t num_pending_events = 0;

            const apr_pollfd_t *next_pending_event;

            apr_interval_time_t poll_timeout;

            int i;

            

            if (APR_RING_EMPTY(&(iom->pending_timeouts), conn_state_t,

                               timeout_list)) {

                poll_timeout = DEFAULT_POLL_TIMEOUT;

            }

            else {

                /* If there are pending timeouts, check whether

                 * any of them have expired.  If none have expired,

                 * use the expiration time on the first one to

                 * determine how long the poll should block.

                 */

                apr_time_t now = apr_time_now();

                conn_state_t *cs = APR_RING_FIRST(&(iom->pending_timeouts));

                if (cs->expiration_time <= now) {

                    do {

                        APR_RING_REMOVE(cs, timeout_list);

                        apr_pollset_remove(iom->pollset, &(cs->pfd));

                        APR_RING_INSERT_TAIL(&(iom->expired_timeouts), cs,

                                             conn_state_t, timeout_list);

                        if (APR_RING_EMPTY(&(iom->pending_timeouts),

                                           conn_state_t, timeout_list)) {

                            break;

                        }

                        cs = APR_RING_FIRST(&(iom->pending_timeouts));

                    } while (cs->expiration_time <= now);

                    continue;

                }

                else {

                    poll_timeout = cs->expiration_time - now;

                }

            }



            apr_thread_mutex_unlock(iom->lock);



            rv = apr_pollset_poll(iom->pollset, poll_timeout,

                                  &num_pending_events, &next_pending_event);



            if ((rv != APR_SUCCESS) && !APR_STATUS_IS_TIMEUP(rv) && !APR_STATUS_IS_EINTR(rv)) {

                apr_thread_mutex_unlock(iom->pollset_lock);

                return rv;

            }

            apr_thread_mutex_lock(iom->lock);

            

            if (num_pending_events > 0) {

                iom->num_pending_events = num_pending_events;

                iom->next_pending_event = next_pending_event;

                for (i = 0; i < num_pending_events; i++) {

                    multiplexable *m = (multiplexable *)next_pending_event[i].client_data;

                    if (m != NULL) {

                        io_multiplexer_remove_internal(iom, m);

                    }

                }

            }

        }

    }

}



apr_status_t io_multiplexer_stop(io_multiplexer *iom, int graceful) {

    iom->stopped = 1;

    return APR_SUCCESS;

}



apr_status_t io_multiplexer_add(io_multiplexer *iom, multiplexable *m,

                                long timeout_in_usec)

{

    apr_status_t rv;

    apr_thread_mutex_lock(iom->lock);

    if (iom->stopped) {

        rv = APR_EINVAL;

    }

    else if (m->type == IOM_CONNECTION) {

        APR_RING_REMOVE(m->c->cs, timeout_list);

        m->c->cs->pfd.client_data = m;

        rv = apr_pollset_add(iom->pollset, &(m->c->cs->pfd));

        if (timeout_in_usec >= 0) {

            /* XXX: Keep the pending_timeouts list sorted */

            m->c->cs->expiration_time = apr_time_now() + timeout_in_usec;

            APR_RING_INSERT_TAIL(&(iom->pending_timeouts), m->c->cs,

                                 conn_state_t, timeout_list);

        }

    }

    else if (m->type == IOM_LISTENER) {

        apr_pollfd_t desc;

        desc.desc_type = APR_POLL_SOCKET;

        desc.desc.s = m->l->sd;

        desc.reqevents = APR_POLLIN;

        desc.client_data = m;

        rv = apr_pollset_add(iom->pollset, &desc);

    }

    else {

        rv = APR_EINVALSOCK;

    }

    apr_thread_mutex_unlock(iom->lock);

    return rv;

}



apr_status_t io_multiplexer_remove(io_multiplexer *iom, multiplexable *m)

{

    apr_status_t rv;

    apr_thread_mutex_lock(iom->lock);

    rv = io_multiplexer_remove_internal(iom, m);

    apr_thread_mutex_unlock(iom->lock);

    return rv;

}



static apr_status_t io_multiplexer_remove_internal(io_multiplexer *iom,

                                                   multiplexable *m)

{

    apr_status_t rv;

    if (iom->stopped) {

        rv = APR_EINVAL;

    }

    else if (m->type == IOM_CONNECTION) {

        APR_RING_REMOVE(m->c->cs, timeout_list);

        rv = apr_pollset_remove(iom->pollset, &(m->c->cs->pfd));

    }

    else if (m->type == IOM_LISTENER) {

        apr_pollfd_t desc;

        desc.desc_type = APR_POLL_SOCKET;

        desc.desc.s = m->l->sd;

        desc.reqevents = APR_POLLIN;

        desc.client_data = NULL;

        rv = apr_pollset_remove(iom->pollset, &desc);

    }

    else {

        rv = APR_EINVALSOCK;

    }

    return rv;

}

server/mpm/experimental/leader/io_multiplexer.h

0 → 100644
+58 −0
Original line number Diff line number Diff line 
/* Copyright 2005 The Apache Software Foundation or its licensors, as

 * applicable.

 *

 * Licensed 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.

 */



#ifndef APACHE_MPM_EVENT_IOMUX_H

#define APACHE_MPM_EVENT_IOMUX_H



#include "apr_pools.h"

#include "apr_network_io.h"

#include "apr_poll.h"



#include "ap_listen.h"

#include "httpd.h"



typedef struct io_multiplexer io_multiplexer;



typedef struct {

    enum { IOM_LISTENER, IOM_CONNECTION } type;

    union {

        ap_listen_rec *l;

        conn_rec *c;

    };

} multiplexable;



/* Flag to set in apr_pollfd_t.rtnevents upon timeout

 * XXX: Find a way to make sure this never collides with any value

 *      set by APR

 */

#define IOM_POLL_TIMEOUT 0x8000



apr_status_t io_multiplexer_create(io_multiplexer **iom, apr_pool_t *p,

                                   apr_uint32_t max_descriptors);



apr_status_t io_multiplexer_get_event(io_multiplexer *iom, apr_pollfd_t *event);



apr_status_t io_multiplexer_stop(io_multiplexer *iom, int graceful);



#define IOM_TIMEOUT_INFINITE -1



apr_status_t io_multiplexer_add(io_multiplexer *iom, multiplexable *m,

                                long timeout_in_usec);



apr_status_t io_multiplexer_remove(io_multiplexer *iom, multiplexable *m);



#endif /* APACHE_MPM_EVENT_IOMUX_H */

server/mpm/experimental/leader/leader.c

+180 −393

File changed.

Preview size limit exceeded, changes collapsed.