src/http/modules/mxcache/ngx_connection_pool.c

Go to the documentation of this file.

/*
 * Copyright (C) Igor Sysoev
 */
/*
 *  Copyright 2009 DreamLab Onet.pl Sp. z o.o.
 *  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.
 *
 *   Authors:
 *   Michal Szklarzewicz <michal(x)szklarzewicz(y)gmail(x)com>
 *   Grzegorz Jablonski  <jablonski(x)grzegorz(y)gmail(x)com>
 *   (use dot instead of x, and @ instead of y)
 */
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_http.h>
#include <nginx.h>
#include <ngx_log.h>

#include "ngx_connection_pool.h"

#if (NGX_THREADS)
#error "This module is not thread safe"
#endif

ngx_atomic_t    connecton_pool_pcounter = 1;
ngx_atomic_t    *ngx_connecton_pool_pcounter = &connecton_pool_pcounter;

/* create new connection pool */
ngx_connection_pool_t*
ngx_connection_pool_create(ngx_log_t *_log, struct sockaddr_in *_sockaddr,
                ngx_uint_t _sockaddr_size, ngx_uint_t _stored_connections_limit)
{
        ngx_connection_pool_t *connection_pool;
        ngx_pool_t *pool;
        char _name[256]; // zdecydowanie char a nie u_char, innaczej sie nie kompiluje

        pool = ngx_create_pool(1024, _log);
        connection_pool = ngx_pcalloc(pool, sizeof(ngx_connection_pool_t));

        //assign
        connection_pool->pool = pool;
        connection_pool->log = _log;
        connection_pool->sockaddr = _sockaddr;
        connection_pool->sockaddr_size = _sockaddr_size;
        connection_pool->stored_connections_limit = _stored_connections_limit;
        connection_pool->stored_connections = 0;
        connection_pool->returned_connections = 0;

        connection_pool->first_connection = 0;
        connection_pool->last_connection = 0;

        //name
        //sprintf(_name, "connection_from_pool_%d", ngx_atomic_fetch_add(ngx_connecton_pool_pcounter, 1));
        ngx_snprintf( (u_char*)_name, sizeof(_name)/sizeof(_name[0]), 
                                "connection_from_pool_%d",
                                ngx_atomic_fetch_add(ngx_connecton_pool_pcounter, 1));
        connection_pool->name.len = strlen(_name);
        connection_pool->name.data = ngx_pcalloc(pool, connection_pool->name.len);
        ngx_memcpy(connection_pool->name.data, _name, connection_pool->name.len);

        //pool created, return it
        return connection_pool;
}

static void
ngx_connection_pool_strip_broken(ngx_connection_pool_t *pool)
{
        ngx_err_t err;
        ngx_int_t rc;
        ngx_pooled_connection_t *current;
        #define testbufflen 5
        u_char testbuff[testbufflen];

        while (pool->first_connection != 0) {
                current = pool->first_connection;
                rc = read(current->fd, testbuff, testbufflen);
                err = ngx_errno;
                ngx_log_debug2(NGX_LOG_DEBUG, pool->log, errno,
                                "[ngx_connection_pool_strip_broken] FD = %d, RET = %d",
                                current->fd, rc);

                if (rc == -1 || rc == 0) {
                        //err = ngx_errno; /MS: saving ngx_errno there is pointless, it could be already changed by log_debug3
                        if (rc == -1 && err == EAGAIN) {
                                break; //this one was OK (most likely)
                        } else {
                                //this one was broken
                                pool->first_connection = current->next;
                                if (pool->first_connection != 0) {
                                        pool->first_connection->previous = 0;
                                } else if (pool->last_connection == current) {
                                        pool->last_connection = 0;
                                }

                                //delete current
                                //ngx_close_connection(current->connection);
                                //do not call ngx_close_connection as current->connection == 0

                                if (ngx_close_socket(current->fd) == -1) {
                                        ngx_log_debug1(NGX_LOG_DEBUG, pool->log, ngx_errno,
                                                "[ngx_connection_pool_strip_broken] %d failed", current->fd);
                                }

                                ngx_destroy_pool(current->pool);

                                pool->stored_connections--;
                                if(rc == -1) {
                                        ngx_log_debug0(NGX_LOG_DEBUG, pool->log, err, 
                                                "[ngx_connection_pool_strip_broken] Free failed connection");
                                } else {
                                        ngx_log_debug0(NGX_LOG_DEBUG, pool->log, 0,
                                                "[ngx_connection_pool_strip_broken] Free connection, closed by foreign host");
                                }
                        }
                } else {
                        break; //this one was OK
                }
        }
}

static ngx_int_t
ngx_connection_pool_get_new(ngx_connection_pool_t* pool,
                ngx_pooled_connection_t** new_connection)
{
        ngx_peer_connection_t peer;
        ngx_pool_t *allocator_pool;
        ngx_int_t rc;
        ngx_err_t err;
        
        if(pool->stored_connections >= pool->stored_connections_limit) {
                ngx_log_error(NGX_LOG_ERR, pool->log, 0, "[ngx_connection_pool_get_new] Connection limit hit. Consider increasing it.");
                return NGX_ERROR;
        }
        
        //okey, connect now

        peer.sockaddr = (struct sockaddr*) pool->sockaddr;
        peer.socklen = pool->sockaddr_size;
        peer.name = &pool->name;
        peer.get = ngx_event_get_peer;
        peer.log = pool->log;
        peer.log_error = NGX_ERROR_ERR;

        rc = ngx_event_connect_peer(&peer);

        if (rc == NGX_ERROR || rc == NGX_BUSY || rc == NGX_DECLINED) {
                //(rc == NGX_ERROR) ? err = ngx_errno : err = 0;
                if(rc == NGX_ERROR) {
                        err = ngx_errno;
                } else {
                        err = 0;
                }
                ngx_log_error(NGX_LOG_ERR, pool->log, err,
                                "[ngx_connection_pool_get_new] Connect failed, error was %d", rc);
                if (peer.connection != NULL && peer.connection->fd != -1) {
                        ngx_close_connection(peer.connection);
                }

                if (rc == NGX_ERROR) {
                        return NGX_ERROR;
                } else {
                        return NGX_AGAIN;
                }
        }

        allocator_pool = ngx_create_pool(256, pool->log);
        (*new_connection) = ngx_pcalloc(allocator_pool, sizeof(ngx_pooled_connection_t));

        (*new_connection)->fd = peer.connection->fd;
        (*new_connection)->connection = peer.connection;

        (*new_connection)->owner_pool = pool;
        (*new_connection)->pool = allocator_pool;
        (*new_connection)->next = 0;
        (*new_connection)->previous = 0;

        (*new_connection)->connection->data = *new_connection;
        (*new_connection)->data = 0;

        pool->stored_connections++;
        pool->returned_connections++;

        ngx_log_debug1(NGX_LOG_DEBUG, pool->log, 0,
                        "[ngx_connection_pool_get_new] Success, FD = %d",
                        (*new_connection)->connection->fd);
        return NGX_OK;
}

ngx_int_t
ngx_connection_pool_get(ngx_connection_pool_t* pool,
                ngx_pooled_connection_t** new_connection, ngx_int_t force_new)
{
        ngx_int_t event;
        ngx_socket_t s;
        ngx_event_t *rev, *wev;
        ngx_connection_t *c;

        if (force_new == 1) {
                return ngx_connection_pool_get_new(pool, new_connection);
        }

        ngx_connection_pool_strip_broken(pool); /* czy potrzebujemy tego zawsze ? - tak, bezwzglednie */

        //check if there are some stored connections
        if (pool->first_connection != NULL) {
                //return old connection
                (*new_connection) = pool->first_connection;
                pool->first_connection = (*new_connection)->next;
                if (pool->first_connection != 0) {
                        pool->first_connection->previous = 0;
                } else if (pool->last_connection == (*new_connection)) {
                        pool->last_connection = 0;
                }
                (*new_connection)->next = 0;
                (*new_connection)->previous = 0;

                //allocated nginx-like connection for (*new_connection)->fd
                s = (*new_connection)->fd;

                c = ngx_get_connection(s, pool->log);
                if (c == NULL) {
                        if (ngx_close_socket(s) == -1) {
                                ngx_log_error(NGX_LOG_ERR, pool->log, ngx_socket_errno,
                                        "[ngx_connection_pool_get] ngx_close_socket_n failed");
                        }
                ngx_log_debug3(NGX_LOG_DEBUG, pool->log, 0,
                        "[ngx_connection_pool_get] Return old, get_connection failed.\
 Could hit nginx used connections limit! FD = %d, PoolCount = %d %d",
                        s, pool->stored_connections, pool->returned_connections);

                //destroy *new_connection

                ngx_destroy_pool((*new_connection)->pool);

                pool->returned_connections--;
                pool->stored_connections--;
                ngx_log_debug3(NGX_LOG_DEBUG,pool->log,0,"[ngx_connection_pool_release] Connection forcefully released, FD = %d, PoolCount = %d %d", s, pool->stored_connections, pool->returned_connections);

                return NGX_ERROR;
                }

                c->recv = ngx_recv;
                c->send = ngx_send;
                c->recv_chain = ngx_recv_chain;
                c->send_chain = ngx_send_chain;

                c->sendfile = 1;

                c->log_error = NGX_ERROR_ERR;

                if (((struct sockaddr*)pool->sockaddr)->sa_family != AF_INET) {
                        c->tcp_nopush = NGX_TCP_NOPUSH_DISABLED;
                        c->tcp_nodelay = NGX_TCP_NODELAY_DISABLED;

#if (NGX_SOLARIS)
                        c->sendfile = 0;
#endif
                }

                rev = c->read;
                wev = c->write;

                rev->log = pool->log;
                wev->log = pool->log;

                (*new_connection)->connection = c;

                if (ngx_add_conn) {
                        if (ngx_add_conn(c) == NGX_ERROR) {
                                ngx_log_error(NGX_LOG_ERR, pool->log, 0,
                                        "[ngx_connection_pool_release] ngx_add_conn(c) failed, FD = %d, PoolCount = %d %d",
                                        s, pool->stored_connections, pool->returned_connections);
                                ngx_connection_pool_release(*new_connection);
                                return NGX_ERROR;
                        }
                }

                if (ngx_add_conn) {
                        wev->ready = 1;
                } else if (ngx_event_flags & NGX_USE_AIO_EVENT) {
                        rev->ready = 1;
                        wev->ready = 1;
                } else {
                        if (ngx_event_flags & NGX_USE_CLEAR_EVENT) {
                                event = NGX_CLEAR_EVENT;
                        } else {
                                event = NGX_LEVEL_EVENT;
                        }

                        if (ngx_add_event(rev, NGX_READ_EVENT, event) != NGX_OK) {
                        //return NGX_ERROR;
                                ngx_log_error(NGX_LOG_ERR, pool->log, 0,
                                        "[ngx_connection_pool_release] ngx_add_event(rev, NGX_READ_EVENT, event) failed, FD = %d, PoolCount = %d %d",
                                        s, pool->stored_connections, pool->returned_connections);
                                ngx_connection_pool_release(*new_connection);
                        return NGX_ERROR;
                        }

                        wev->ready = 1;
                }
        //OK
                (*new_connection)->connection->data = *new_connection;
                pool->returned_connections++;
                ngx_log_debug3(NGX_LOG_DEBUG,pool->log, 0, 
                        "[ngx_connection_pool_get] Return old, FD = %d, PoolCount = %d %d",
                        (*new_connection)->connection->fd, pool->stored_connections, pool->returned_connections);
                return NGX_OK; //ok
        }
//get new connection
//return ngx_connection_pool_get_new(pool);
//new_connection
        ngx_log_debug2(NGX_LOG_DEBUG, pool->log, 0,
                "[ngx_connection_pool_get] Allocate new, PoolCount = %d %d",
                pool->stored_connections, pool->returned_connections);
        return ngx_connection_pool_get_new(pool, new_connection);
}

void
ngx_connection_pool_return(ngx_pooled_connection_t *connection_context)
{
        ngx_connection_t *connection;
        ngx_connection_pool_t *pool;
        ngx_uint_t log_error;//, level;
        ngx_socket_t fd;

        connection = connection_context->connection;
        connection_context->data = 0;

        //okey, connection is being returned
        pool = connection_context->owner_pool;

        //free connection

        fd = connection_context->connection->fd;
        if (fd != connection_context->fd) {
                ngx_log_error(NGX_LOG_ERR, connection->log, 0,
                                "[ngx_connection_pool_return] Invalid FD %d != %d", fd,
                                connection_context->fd);
        }

        if (fd == -1) {
                ngx_connection_pool_release(connection_context); //error should be handled there
                return;
        }

        if (connection->read->timer_set) {
                ngx_del_timer(connection->read);
        }

        if (connection->write->timer_set) {
                ngx_del_timer(connection->write);
        }

        if (ngx_del_conn) {
                //ngx_del_conn(connection, NGX_CLOSE_EVENT);
                ngx_del_conn(connection, 0);
        } else {
                if (connection->read->active || connection->read->disabled) {
                        ngx_del_event(connection->read, NGX_READ_EVENT, NGX_CLOSE_EVENT); //XXX?? NGX_CLOSE_EVENT -> 0
                }

                if (connection->write->active || connection->write->disabled) {
                        ngx_del_event(connection->write, NGX_WRITE_EVENT, NGX_CLOSE_EVENT); //XXX?? NGX_CLOSE_EVENT -> 0
                }
        }

#if (NGX_THREADS)

        ngx_mutex_lock(ngx_posted_events_mutex);

        if (connection->read->prev) {
                ngx_delete_posted_event(connection->read);
        }

        if (connection->write->prev) {
                ngx_delete_posted_event(connection->write);
        }

        connection->read->closed = 1;
        connection->write->closed = 1;

        if (connection->single_connection) {
                ngx_unlock(&connection->lock);
                connection->read->locked = 0;
                connection->write->locked = 0;
        }

        ngx_mutex_unlock(ngx_posted_events_mutex);

#else

        if (connection->read->prev) {
                ngx_delete_posted_event(connection->read);
        }

        if (connection->write->prev) {
                ngx_delete_posted_event(connection->write);
        }

        connection->read->closed = 1;
        connection->write->closed = 1;

#endif

        log_error = connection->log_error;

        ngx_free_connection(connection);

        // fd = connection->fd;
        connection->fd = (ngx_socket_t) -1;

        //connection is freed from nginx inner structs now, only connection_context->fd remains
        connection_context->connection = NULL;

        if (pool->first_connection == NULL) {
                //first returned
                pool->first_connection = connection_context;
                pool->last_connection = connection_context;

                connection_context->next = NULL;
                connection_context->previous = NULL; //released
        } else {// push at end, multiple returned exists
                connection_context->next = NULL;
                connection_context->previous = pool->last_connection;
                pool->last_connection->next = connection_context;
                pool->last_connection = connection_context;
        }
        pool->returned_connections--;
        ngx_log_debug3( NGX_LOG_DEBUG, pool->log, 0,
                        "[ngx_connection_pool_return] Connection returned to pool, FD = %d, PoolCount = %d %d",
                        connection_context->fd, pool->stored_connections,
                        pool->returned_connections);
}

void ngx_connection_pool_release(ngx_pooled_connection_t *connection_context)
{
        ngx_connection_t *connection;
        ngx_connection_pool_t *pool;

        connection = connection_context->connection;
        //okey, connection is being returned
        pool = connection_context->owner_pool;

        //delete current
        ngx_close_connection(connection_context->connection);
        ngx_destroy_pool(connection_context->pool);

        pool->returned_connections--;
        pool->stored_connections--;
        ngx_log_debug3( NGX_LOG_DEBUG, pool->log, 0,
                        "[ngx_connection_pool_release] Connection forcefully released, FD = %d, PoolCount = %d %d",
                        connection->fd, pool->stored_connections,
                        pool->returned_connections);
}

---