You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

1095 lines
34 KiB

#include <lighttpd/backends.h>
typedef struct liBackendConnection_p liBackendConnection_p;
typedef struct liBackendWorkerPool liBackendWorkerPool;
typedef struct liBackendPool_p liBackendPool_p;
struct liBackendWait {
li_tstamp ts_started;
/* 3 different states:
* - con != NULL: connection associated (may need to move between threads/...)
* - failed = TRUE: backend is down
* - queued in wait_queue with link
*/
liBackendConnection_p *con;
GList wait_queue_link; /* link in wait_queue (either pool or wpool) */
gboolean failed;
liVRequest *vr;
liJobRef *vr_ref;
};
/* members marked with "[pool]" are protected by the pool lock, the others belong to the worker */
struct liBackendConnection_p {
liBackendConnection public;
liBackendPool_p *pool;
gint ndx; /* [pool] index in the worker pool (worker ? worker : worker_next )->connections */
gboolean active; /* [pool] */
gint requests;
liWaitQueueElem timeout_elem; /* ([pool]) idle or connect */
/* if != NULL the connection is reserved by a vrequest,
* and waiting to get transferred to the correct worker
*/
liBackendWait *wait; /* [pool] */
/* worker != NULL: connection belongs to worker
* worker_next != NULL: worker should release connection, and wakeup worker_next
* worker == NULL:
* worker_next must be != NULL: worker_next is waiting to be woken up, but
* if the connection isn't reserved anymore (wait == NULL) another worker can
* steal it
* changing worker leads to the corresponding attach/detach callbacks
*/
liWorker *worker, *worker_next; /* [pool] */
};
/* members marked with "[pool]" are protected by the pool lock, the others belong to the worker */
struct liBackendWorkerPool {
liBackendPool_p *pool;
liWorker *wrk;
liEventAsync wakeup;
guint active, reserved, idle, pending; /* [pool] connection counts */
/* attached connections (con->worker != NULL) may only be removed (and added) by the owning worker */
GPtrArray *connections; /* [pool] <liBackendConnection_p> ordered: active, reserved, idle */
liWaitQueue idle_queue; /* <liBackendConnection_p> */
liWaitQueue connect_queue; /* <liBackendConnection_p> pending connects */
/* waiting vrequests: per worker queue if there is no connection limit (<= 0) */
GQueue wait_queue; /* <liBackendWait> */
liEventTimer wait_queue_timer;
gboolean initialized; /* [pool] only interesting if pool->initialized is false */
};
struct liBackendPool_p {
liBackendPool public;
GMutex *lock;
liBackendWorkerPool *worker_pools;
guint active, reserved, idle, pending, total; /* [pool] connection counts. */
/* waiting vrequests: global queue if connection limit > 0 is used */
GQueue wait_queue; /* <liBackendWait> */
/* ^^ should use timer in worker-pool? */
li_tstamp ts_disabled_till;
gboolean initialized, shutdown;
};
static void S_backend_pool_distribute(liBackendPool_p *pool, liWorker *wrk);
static void backend_con_watch_for_close_cb(liEventBase *watcher, int events);
static void _call_thread_cb(liBackendConnectionThreadCB cb, liBackendPool *bpool, liWorker *wrk, liBackendConnection *bcon) {
if (cb) cb(bpool, wrk, bcon);
}
#define BACKEND_THREAD_CB(name, pool, wrk, con) _call_thread_cb(pool->public.config->callbacks->name##_cb, &pool->public, wrk, &con->public)
static void S_backend_pool_worker_remove_con(liBackendPool_p *pool, liBackendConnection_p *con) {
liWorker *cur = con->worker != NULL ? con->worker : con->worker_next;
liBackendWorkerPool *wpool = &pool->worker_pools[cur->ndx];
gint ndx = con->ndx;
gint last_active_ndx = wpool->active - 1;
gint last_reserved_ndx = last_active_ndx + wpool->reserved;
gint last_idle_ndx = last_reserved_ndx + wpool->idle;
LI_FORCE_ASSERT(con->ndx >= 0 && (guint)con->ndx < wpool->connections->len);
LI_FORCE_ASSERT(g_ptr_array_index(wpool->connections, con->ndx) == con);
LI_FORCE_ASSERT(last_idle_ndx == (gint) wpool->connections->len - 1);
if (ndx <= last_active_ndx) {
--wpool->active;
--pool->active;
} else if (ndx <= last_reserved_ndx) {
--wpool->reserved;
--pool->reserved;
} else {
--wpool->idle;
--pool->idle;
}
--pool->total;
if (ndx < last_active_ndx) {
liBackendConnection_p *move = g_ptr_array_index(wpool->connections, last_active_ndx);
g_ptr_array_index(wpool->connections, ndx) = move;
move->ndx = ndx;
ndx = last_active_ndx;
}
if (ndx < last_reserved_ndx) {
liBackendConnection_p *move = g_ptr_array_index(wpool->connections, last_reserved_ndx);
g_ptr_array_index(wpool->connections, ndx) = move;
move->ndx = ndx;
ndx = last_reserved_ndx;
}
if (ndx < last_idle_ndx) {
liBackendConnection_p *move = g_ptr_array_index(wpool->connections, last_idle_ndx);
g_ptr_array_index(wpool->connections, ndx) = move;
move->ndx = ndx;
ndx = last_idle_ndx;
}
g_ptr_array_index(wpool->connections, last_idle_ndx) = NULL;
g_ptr_array_set_size(wpool->connections, last_idle_ndx);
con->ndx = -1;
}
static void S_backend_pool_worker_insert_con(liBackendPool_p *pool, liWorker *wrk, liBackendConnection_p *con) {
liWorker *cur = con->worker != NULL ? con->worker : con->worker_next;
liBackendWorkerPool *wpool;
gint min_ndx, max_ndx;
if (wrk == NULL) wrk = cur;
if (-1 != con->ndx && wrk != cur) {
S_backend_pool_worker_remove_con(pool, con);
}
wpool = &pool->worker_pools[wrk->ndx];
if (-1 == con->ndx) {
con->ndx = wpool->connections->len;
g_ptr_array_add(wpool->connections, con);
++pool->total;
} else if ((guint) con->ndx < wpool->active) {
--wpool->active;
--pool->active;
} else if ((guint) con->ndx < wpool->active + wpool->reserved) {
--wpool->reserved;
--pool->reserved;
} else {
--wpool->idle;
--pool->idle;
}
if (con->active) {
++wpool->active;
++pool->active;
min_ndx = 0;
max_ndx = wpool->active - 1;
} else if (NULL == con->worker || NULL != con->wait) {
++wpool->reserved;
++pool->reserved;
min_ndx = wpool->active;
max_ndx = wpool->active + wpool->reserved - 1;
} else {
++wpool->idle;
++pool->idle;
min_ndx = wpool->active + wpool->reserved;
max_ndx = wpool->active + wpool->reserved + wpool->idle - 1;
}
if (con->ndx < min_ndx) {
if ((guint) con->ndx < wpool->active) {
liBackendConnection_p *move = g_ptr_array_index(wpool->connections, wpool->active);
g_ptr_array_index(wpool->connections, con->ndx) = move;
move->ndx = con->ndx;
con->ndx = wpool->active;
}
if (con->ndx < min_ndx && (guint) con->ndx < wpool->active + wpool->reserved) {
liBackendConnection_p *move = g_ptr_array_index(wpool->connections, wpool->active + wpool->reserved);
g_ptr_array_index(wpool->connections, con->ndx) = move;
move->ndx = con->ndx;
con->ndx = wpool->active + wpool->reserved;
}
g_ptr_array_index(wpool->connections, con->ndx) = con;
LI_FORCE_ASSERT(con->ndx == min_ndx);
} else if (con->ndx > max_ndx) {
if ((guint) con->ndx > wpool->active + wpool->reserved - 1) {
liBackendConnection_p *move = g_ptr_array_index(wpool->connections, wpool->active + wpool->reserved - 1);
g_ptr_array_index(wpool->connections, con->ndx) = move;
move->ndx = con->ndx;
con->ndx = wpool->active + wpool->reserved - 1;
}
if (con->ndx > max_ndx && (guint) con->ndx > wpool->active - 1) {
liBackendConnection_p *move = g_ptr_array_index(wpool->connections, wpool->active - 1);
g_ptr_array_index(wpool->connections, con->ndx) = move;
move->ndx = con->ndx;
con->ndx = wpool->active - 1;
}
g_ptr_array_index(wpool->connections, con->ndx) = con;
LI_FORCE_ASSERT(con->ndx == max_ndx);
} else {
LI_FORCE_ASSERT(con->ndx >= min_ndx && con->ndx <= max_ndx);
}
}
static liBackendConnection_p* backend_connection_new(liBackendWorkerPool *wpool) {
liBackendConnection_p *con = g_slice_new0(liBackendConnection_p);
con->pool = wpool->pool;
con->ndx = -1;
con->worker = wpool->wrk;
return con;
}
static void S_backend_pool_worker_insert_connected(liBackendWorkerPool *wpool, int fd) {
liBackendConnection_p *con = backend_connection_new(wpool);
liBackendPool_p *pool = wpool->pool;
li_event_io_init(&wpool->wrk->loop, "backend connection", &con->public.watcher, NULL, fd, 0);
li_event_set_keep_loop_alive(&con->public.watcher, FALSE);
BACKEND_THREAD_CB(new, pool, wpool->wrk, con);
if (pool->public.config->watch_for_close) {
li_event_set_callback(&con->public.watcher, backend_con_watch_for_close_cb);
li_event_io_set_events(&con->public.watcher, LI_EV_READ);
li_event_start(&con->public.watcher);
}
li_waitqueue_push(&wpool->idle_queue, &con->timeout_elem);
S_backend_pool_worker_insert_con(pool, NULL, con);
}
static void S_backend_pool_failed(liBackendWorkerPool *wpool) {
liBackendPool_p *pool = wpool->pool;
GList *elem;
if (pool->public.config->disable_time > 0) {
pool->ts_disabled_till = li_cur_ts(wpool->wrk) + pool->public.config->disable_time;
}
while (NULL != (elem = g_queue_pop_head_link(&pool->wait_queue))) {
liBackendWait *bwait = LI_CONTAINER_OF(elem, liBackendWait, wait_queue_link);
bwait->failed = TRUE;
li_job_async(bwait->vr_ref);
}
for (guint i = 0, len = wpool->wrk->srv->worker_count; i < len; ++i) {
liBackendWorkerPool *_wpool = &pool->worker_pools[i];
while (NULL != (elem = g_queue_pop_head_link(&_wpool->wait_queue))) {
liBackendWait *bwait = LI_CONTAINER_OF(elem, liBackendWait, wait_queue_link);
bwait->failed = TRUE;
li_job_async(bwait->vr_ref);
}
}
}
/* See http://www.cyberconf.org/~cynbe/ref/nonblocking-connects.html
* for a discussion on async connects
*/
static void backend_pool_worker_connect_timeout(liWaitQueue *wq, gpointer data) {
liBackendWorkerPool *wpool = data;
liBackendPool_p *pool = wpool->pool;
liWaitQueueElem *elem;
const liBackendConfig *config = pool->public.config;
liServer *srv = wpool->wrk->srv;
g_mutex_lock(pool->lock);
while (NULL != (elem = li_waitqueue_pop(wq))) {
liBackendConnection_p *con = LI_CONTAINER_OF(elem, liBackendConnection_p, timeout_elem);
li_event_clear(&con->public.watcher);
close(li_event_io_fd(&con->public.watcher));
g_slice_free(liBackendConnection_p, con);
ERROR(srv, "Couldn't connect to '%s': timeout",
li_sockaddr_to_string(config->sock_addr, wpool->wrk->tmp_str, TRUE)->str);
--wpool->pending;
--wpool->pool->pending;
--wpool->pool->total;
S_backend_pool_failed(wpool);
}
g_mutex_unlock(pool->lock);
li_waitqueue_update(wq);
}
static void backend_con_watch_connect_cb(liEventBase *watcher, int events) {
liEventIO *iowatcher = li_event_io_from(watcher);
liBackendConnection_p *con = LI_CONTAINER_OF(iowatcher, liBackendConnection_p, public.watcher);
liBackendPool_p *pool = con->pool;
const liBackendConfig *config = pool->public.config;
liBackendWorkerPool *wpool = &pool->worker_pools[con->worker->ndx];
liServer *srv = wpool->wrk->srv;
int fd = li_event_io_fd(iowatcher);
struct sockaddr addr;
socklen_t len;
UNUSED(events);
li_event_stop(iowatcher);
li_waitqueue_remove(&wpool->connect_queue, &con->timeout_elem);
g_mutex_lock(pool->lock);
/* Check to see if we can determine our peer's address. */
len = sizeof(addr);
if (getpeername(fd, &addr, &len) == -1) {
/* connect failed; find out why */
int err;
len = sizeof(err);
#ifdef SO_ERROR
if (-1 == getsockopt(fd, SOL_SOCKET, SO_ERROR, (void*)&err, &len)) {
err = errno;
}
#else
{
char ch;
errno = 0;
read(fd, &ch, 1);
err = errno;
}
#endif
ERROR(srv, "Couldn't connect to '%s': %s",
li_sockaddr_to_string(config->sock_addr, wpool->wrk->tmp_str, TRUE)->str,
g_strerror(err));
close(fd);
li_event_clear(iowatcher);
g_slice_free(liBackendConnection_p, con);
--wpool->pending;
--wpool->pool->pending;
--wpool->pool->total;
S_backend_pool_failed(wpool);
} else {
/* connect succeeded */
BACKEND_THREAD_CB(new, pool, wpool->wrk, con);
if (pool->public.config->watch_for_close) {
li_event_set_callback(&con->public.watcher, backend_con_watch_for_close_cb);
li_event_io_set_events(&con->public.watcher, LI_EV_READ);
li_event_start(&con->public.watcher);
}
li_waitqueue_push(&wpool->idle_queue, &con->timeout_elem);
--wpool->pending;
--wpool->pool->pending;
--wpool->pool->total;
S_backend_pool_worker_insert_con(pool, NULL, con);
S_backend_pool_distribute(pool, wpool->wrk);
}
g_mutex_unlock(pool->lock);
}
static void S_backend_pool_worker_insert_pending(liBackendWorkerPool *wpool, int fd) {
liBackendConnection_p *con = backend_connection_new(wpool);
li_event_io_init(&wpool->wrk->loop, "backend connection", &con->public.watcher, backend_con_watch_connect_cb, fd, LI_EV_READ | LI_EV_WRITE);
li_event_set_keep_loop_alive(&con->public.watcher, FALSE);
li_event_start(&con->public.watcher);
++wpool->pending;
++wpool->pool->pending;
++wpool->pool->total;
li_waitqueue_push(&wpool->connect_queue, &con->timeout_elem);
}
static gboolean S_backend_connection_connect(liBackendWorkerPool *wpool) {
const liBackendConfig *config = wpool->pool->public.config;
liServer *srv = wpool->wrk->srv;
int fd;
do {
fd = socket(config->sock_addr.addr->plain.sa_family, SOCK_STREAM, 0);
} while (-1 == fd && errno == EINTR);
if (-1 == fd) {
if (errno == EMFILE) {
li_server_out_of_fds(srv);
}
ERROR(srv, "Couldn't open socket: %s", g_strerror(errno));
return FALSE;
}
li_fd_no_block(fd);
if (-1 == connect(fd, &config->sock_addr.addr->plain, config->sock_addr.len)) {
switch (errno) {
case EINPROGRESS:
case EALREADY:
case EINTR:
S_backend_pool_worker_insert_pending(wpool, fd);
return TRUE;
default:
ERROR(srv, "Couldn't connect to '%s': %s",
li_sockaddr_to_string(config->sock_addr, wpool->wrk->tmp_str, TRUE)->str,
g_strerror(errno));
close(fd);
return FALSE;
}
}
S_backend_pool_worker_insert_connected(wpool, fd);
return TRUE; /* successfully connected */
}
static void S_backend_pool_distribute(liBackendPool_p *pool, liWorker *wrk) {
if (pool->public.config->max_connections <= 0) {
int max_connections = (pool->public.config->max_connections < 0) ? -pool->public.config->max_connections : 128;
liBackendWorkerPool *wpool = &pool->worker_pools[wrk->ndx];
/* only local wpool */
if (0 == wpool->wait_queue.length) return;
while (wpool->wait_queue.length > 0 && wpool->idle > 0) {
liBackendWait *bwait = LI_CONTAINER_OF(g_queue_pop_head_link(&wpool->wait_queue), liBackendWait, wait_queue_link);
liBackendConnection_p *con = g_ptr_array_index(wpool->connections, wpool->active + wpool->reserved);
bwait->con = con;
con->wait = bwait;
con->active = TRUE;
S_backend_pool_worker_insert_con(pool, NULL, con);
li_vrequest_joblist_append(bwait->vr);
}
if (MIN((unsigned int) max_connections, wpool->wait_queue.length) > wpool->pending) {
guint need = MIN((unsigned int) max_connections, wpool->wait_queue.length) - wpool->pending;
for (; need > 0; --need) {
if (!S_backend_connection_connect(wpool)) {
S_backend_pool_failed(wpool);
return;
}
}
/* recursive restart. should only recurse once... */
S_backend_pool_distribute(pool, wrk);
}
} else {
if (0 == pool->wait_queue.length) return;
/* ERROR(wrk->srv, "pool wait queue: %i, pool idle: %i", pool->wait_queue.length, pool->idle); */
if (pool->idle > 0) {
/* distribute backends over all workers:
* + first sort all "bwait"s we're going to assign to idle connections into the worker
* specific queues
* + 1. distribution round: for each worker assign connections from the worker itself
* + 2. distribution round: assign connections from other workers
*/
liServer *srv = wrk->srv;
guint worker_count = srv->worker_count;
guint use = MIN(pool->idle, pool->wait_queue.length);
{
guint i;
for (i = use; i > 0; --i) {
liBackendWait *bwait = LI_CONTAINER_OF(g_queue_pop_head_link(&pool->wait_queue), liBackendWait, wait_queue_link);
liBackendWorkerPool *wpool = &pool->worker_pools[bwait->vr->wrk->ndx];
g_queue_push_tail_link(&wpool->wait_queue, &bwait->wait_queue_link);
}
}
{
guint i;
for (i = 0; i < worker_count; ++i) {
liBackendWorkerPool *wpool = &pool->worker_pools[i];
/* ERROR(wrk->srv, "pool %i: queue: %i, idle: %i", i, wpool->wait_queue.length, wpool->idle); */
while (wpool->wait_queue.length > 0 && wpool->idle > 0) {
liBackendWait *bwait = LI_CONTAINER_OF(g_queue_pop_head_link(&wpool->wait_queue), liBackendWait, wait_queue_link);
liBackendConnection_p *con = g_ptr_array_index(wpool->connections, wpool->active + wpool->reserved);
bwait->con = con;
con->wait = bwait;
con->active = TRUE;
S_backend_pool_worker_insert_con(pool, NULL, con);
if (i == wrk->ndx) {
li_vrequest_joblist_append(bwait->vr);
} else {
li_job_async(bwait->vr_ref);
}
--use;
}
}
}
if (use > 0) {
guint src = 0;
guint i;
LI_FORCE_ASSERT(pool->idle >= use);
for (i = 0; i < worker_count; ++i) {
liBackendWorkerPool *wpool = &pool->worker_pools[i];
while (wpool->wait_queue.length > 0) {
liBackendWait *bwait = LI_CONTAINER_OF(g_queue_pop_head_link(&wpool->wait_queue), liBackendWait, wait_queue_link);
liBackendConnection_p *con;
liBackendWorkerPool *srcpool;
while (0 == pool->worker_pools[src].idle) {
++src;
LI_FORCE_ASSERT(src < worker_count);
}
srcpool = &pool->worker_pools[src];
con = g_ptr_array_index(srcpool->connections, srcpool->active + srcpool->reserved);
bwait->con = con;
con->wait = bwait;
con->worker_next = g_array_index(srv->workers, liWorker*, i);
S_backend_pool_worker_insert_con(pool, NULL, con);
li_event_async_send(&srcpool->wakeup);
}
}
}
}
if (pool->wait_queue.length > pool->pending) {
guint need = MIN((unsigned int) pool->public.config->max_connections - pool->total, pool->wait_queue.length - pool->pending);
if (need > 0) {
liBackendWorkerPool *wpool = &pool->worker_pools[wrk->ndx];
for (; need > 0; --need) {
if (!S_backend_connection_connect(wpool)) {
S_backend_pool_failed(wpool);
return;
}
}
/* recursive restart. should only recurse once... */
S_backend_pool_distribute(pool, wrk);
}
}
}
}
static void S_backend_wait_queue_unshift(GQueue *queue, GList *lnk) {
if (0 == queue->length) {
g_queue_push_head_link(queue, lnk);
} else {
liBackendWait *link_wait = LI_CONTAINER_OF(lnk, liBackendWait, wait_queue_link);
GList *cursor = queue->head;
liBackendWait *bwait = LI_CONTAINER_OF(cursor, liBackendWait, wait_queue_link);
if (bwait->ts_started > link_wait->ts_started) {
g_queue_push_head_link(queue, lnk);
return;
}
do {
cursor = cursor->next;
if (NULL == cursor) {
g_queue_push_tail_link(queue, lnk);
return;
}
bwait = LI_CONTAINER_OF(cursor, liBackendWait, wait_queue_link);
} while (bwait->ts_started < link_wait->ts_started);
/* insert lnk before cursor; lnk will neither be the first nor the last element,
* so we don't have to udpate queue->head/tail
*/
lnk->next = cursor;
lnk->prev = cursor->prev;
cursor->prev->next = lnk;
cursor->prev = lnk;
}
}
static void backend_connection_close(liBackendPool_p *pool, liBackendConnection_p *con, gboolean have_lock) {
liWorker *wrk = con->worker; /* only close attached connections */
liBackendWorkerPool *wpool = &pool->worker_pools[wrk->ndx];
int fd;
if (!have_lock) g_mutex_lock(pool->lock);
S_backend_pool_worker_remove_con(pool, con);
if (NULL != con->wait) {
con->wait->con = NULL;
if (pool->public.config->max_connections <= 0) {
S_backend_wait_queue_unshift(&wpool->wait_queue, &con->wait->wait_queue_link);
} else {
S_backend_wait_queue_unshift(&pool->wait_queue, &con->wait->wait_queue_link);
}
S_backend_pool_distribute(pool, con->wait->vr->wrk);
con->wait = NULL;
}
if (!have_lock) g_mutex_unlock(pool->lock);
/* don't need lock here */
li_waitqueue_remove(&wpool->idle_queue, &con->timeout_elem);
BACKEND_THREAD_CB(close, pool, wrk, con);
fd = li_event_io_fd(&con->public.watcher);
li_event_clear(&con->public.watcher);
if (-1 != fd) li_worker_add_closing_socket(wrk, fd);
g_slice_free(liBackendConnection_p, con);
}
static void backend_con_watch_for_close_cb(liEventBase *watcher, int events) {
liEventIO *iowatcher = li_event_io_from(watcher);
liBackendConnection_p *con = LI_CONTAINER_OF(iowatcher, liBackendConnection_p, public.watcher);
liBackendPool_p *pool = con->pool;
char c;
int r;
UNUSED(events);
r = read(li_event_io_fd(iowatcher), &c, 1);
if (-1 == r && (EAGAIN == errno || EWOULDBLOCK == errno || EINTR == errno)) return;
/* TODO: log error when read data */
backend_connection_close(pool, con, FALSE);
}
static void backend_pool_worker_run_reserved(liBackendWorkerPool *wpool) {
liBackendPool_p *pool = wpool->pool;
liWorker *wrk = wpool->wrk;
g_mutex_lock(pool->lock);
while (wpool->reserved > 0) {
liBackendConnection_p *con = g_ptr_array_index(wpool->connections, wpool->active);
if (NULL == con->worker) {
/* attach */
LI_FORCE_ASSERT(con->worker_next == wrk);
con->worker = wrk;
con->worker_next = NULL;
li_event_attach(&wrk->loop, &con->public.watcher);
li_waitqueue_push(&wpool->idle_queue, &con->timeout_elem);
BACKEND_THREAD_CB(attach_thread, pool, wrk, con);
if (-1 == li_event_io_fd(&con->public.watcher)) {
backend_connection_close(pool, con, TRUE);
continue;
}
if (NULL == con->wait) {
S_backend_pool_worker_insert_con(pool, wrk, con);
S_backend_pool_distribute(pool, wrk);
continue;
}
}
LI_FORCE_ASSERT(NULL != con->wait);
if (NULL == con->worker_next) {
LI_FORCE_ASSERT(con->wait->vr->wrk == wrk);
if (!con->active) {
con->active = TRUE;
S_backend_pool_worker_insert_con(pool, wrk, con);
li_vrequest_joblist_append(con->wait->vr);
}
continue;
} else {
/* detach and send to target worker */
BACKEND_THREAD_CB(detach_thread, pool, wrk, con);
if (-1 == li_event_io_fd(&con->public.watcher)) {
backend_connection_close(pool, con, TRUE);
continue;
}
li_event_detach(&con->public.watcher);
li_waitqueue_remove(&wpool->idle_queue, &con->timeout_elem);
S_backend_pool_worker_insert_con(pool, con->worker_next, con);
con->worker = NULL;
li_event_async_send(&pool->worker_pools[con->worker_next->ndx].wakeup);
}
}
g_mutex_unlock(pool->lock);
}
static void backend_pool_worker_run(liEventBase *watcher, int events) {
liEventAsync *async_watcher = li_event_async_from(watcher);
liBackendWorkerPool *wpool = LI_CONTAINER_OF(async_watcher, liBackendWorkerPool, wakeup);
UNUSED(events);
backend_pool_worker_run_reserved(wpool);
}
static void backend_pool_worker_idle_timeout(liWaitQueue *wq, gpointer data) {
liBackendWorkerPool *wpool = data;
liBackendPool_p *pool = wpool->pool;
liWaitQueueElem *elem;
backend_pool_worker_run_reserved(wpool);
while (NULL != (elem = li_waitqueue_pop(wq))) {
liBackendConnection_p *con = LI_CONTAINER_OF(elem, liBackendConnection_p, timeout_elem);
backend_connection_close(pool, con, FALSE);
}
li_waitqueue_update(wq);
}
static void S_backend_pool_update_wait_queue_timer(liBackendWorkerPool *wpool) {
liBackendPool_p *pool = wpool->pool;
if (pool->wait_queue.length > 0) {
li_tstamp now = li_cur_ts(wpool->wrk);
liBackendWait *bwait = LI_CONTAINER_OF(g_queue_peek_head_link(&pool->wait_queue), liBackendWait, wait_queue_link);
li_tstamp repeat = bwait->ts_started + pool->public.config->wait_timeout - now;
if (repeat < 0.05) repeat = 0.05;
li_event_timer_once(&wpool->wait_queue_timer, repeat);
} else {
/* stop timer if queue empty */
li_event_stop(&wpool->wait_queue_timer);
}
}
static void backend_pool_wait_queue_timeout(liEventBase *watcher, int events) {
liBackendWorkerPool *wpool = LI_CONTAINER_OF(li_event_timer_from(watcher), liBackendWorkerPool, wait_queue_timer);
liBackendPool_p *pool = wpool->pool;
li_tstamp due = li_cur_ts(wpool->wrk) - pool->public.config->wait_timeout;
UNUSED(events);
g_mutex_lock(pool->lock);
while (pool->wait_queue.length > 0) {
liBackendWait *bwait = LI_CONTAINER_OF(g_queue_peek_head_link(&pool->wait_queue), liBackendWait, wait_queue_link);
if (bwait->ts_started <= due) {
g_queue_pop_head_link(&pool->wait_queue);
bwait->failed = TRUE;
li_job_async(bwait->vr_ref);
} else {
break;
}
}
S_backend_pool_update_wait_queue_timer(wpool);
g_mutex_unlock(pool->lock);
}
static gpointer backend_pool_worker_init(liWorker *wrk, gpointer fdata) {
liBackendPool_p *pool = fdata;
liBackendWorkerPool *wpool = &pool->worker_pools[wrk->ndx];
guint idle_timeout = pool->public.config->idle_timeout;
if (wpool->initialized) return NULL;
li_event_async_init(&wrk->loop, "backend manager", &wpool->wakeup, backend_pool_worker_run);
if (idle_timeout < 1) idle_timeout = 5;
li_waitqueue_init(&wpool->idle_queue, &wrk->loop, "backend idle queue", backend_pool_worker_idle_timeout, idle_timeout, wpool);
li_waitqueue_init(&wpool->connect_queue, &wrk->loop, "backend connect queue", backend_pool_worker_connect_timeout, pool->public.config->connect_timeout, wpool);
li_event_timer_init(&wrk->loop, "backend wait timeout", &wpool->wait_queue_timer, backend_pool_wait_queue_timeout);
li_event_set_keep_loop_alive(&wpool->wait_queue_timer, FALSE);
wpool->initialized = TRUE;
return NULL;
}
static void backend_pool_worker_init_done(gpointer cbdata, gpointer fdata, GPtrArray *result, gboolean complete) {
liBackendPool_p *pool = fdata;
UNUSED(cbdata);
UNUSED(result);
UNUSED(complete);
pool->initialized = TRUE;
}
static void S_backend_pool_init(liWorker *wrk, liBackendPool_p *pool) {
LI_FORCE_ASSERT(!pool->shutdown);
if (pool->initialized) return;
if (pool->worker_pools == NULL) {
guint i, l = wrk->srv->worker_count;
pool->worker_pools = g_slice_alloc0(sizeof(liBackendWorkerPool) * l);
for (i = 0; i < l; ++i) {
liBackendWorkerPool *wpool = &pool->worker_pools[i];
wpool->wrk = g_array_index(wrk->srv->workers, liWorker*, i);
wpool->pool = pool;
wpool->connections = g_ptr_array_new();
}
li_collect_start(wrk, backend_pool_worker_init, pool, backend_pool_worker_init_done, NULL);
}
backend_pool_worker_init(wrk, pool);
}
static gpointer backend_pool_worker_shutdown(liWorker *wrk, gpointer fdata) {
liBackendPool_p *pool = fdata;
liBackendWorkerPool *wpool = &pool->worker_pools[wrk->ndx];
liWaitQueueElem *elem;
backend_pool_worker_run_reserved(wpool);
li_event_clear(&wpool->wakeup);
li_event_clear(&wpool->wait_queue_timer);
g_mutex_lock(pool->lock);
while (NULL != (elem = li_waitqueue_pop_force(&wpool->idle_queue))) {
liBackendConnection_p *con = LI_CONTAINER_OF(elem, liBackendConnection_p, timeout_elem);
backend_connection_close(pool, con, TRUE);
}
li_waitqueue_stop(&wpool->idle_queue);
while (NULL != (elem = li_waitqueue_pop_force(&wpool->connect_queue))) {
liBackendConnection_p *con = LI_CONTAINER_OF(elem, liBackendConnection_p, timeout_elem);
int fd = li_event_io_fd(&con->public.watcher);
li_event_clear(&con->public.watcher);
close(fd);
g_slice_free(liBackendConnection_p, con);
--wpool->pending;
--wpool->pool->pending;
--wpool->pool->total;
}
li_waitqueue_stop(&wpool->connect_queue);
LI_FORCE_ASSERT(0 == wpool->active);
LI_FORCE_ASSERT(0 == wpool->reserved);
LI_FORCE_ASSERT(0 == wpool->idle);
LI_FORCE_ASSERT(0 == wpool->pending);
g_ptr_array_free(wpool->connections, TRUE);
g_mutex_unlock(pool->lock);
return NULL;
}
static void backend_pool_worker_shutdown_done(gpointer cbdata, gpointer fdata, GPtrArray *result, gboolean complete) {
liBackendPool_p *pool = fdata;
UNUSED(cbdata);
UNUSED(result);
UNUSED(complete);
pool->public.config->callbacks->free_cb(&pool->public);
if (pool->worker_pools != NULL) {
liServer *srv = pool->worker_pools[0].wrk->srv;
g_slice_free1(sizeof(liBackendPool_p) * srv->worker_count, pool->worker_pools);
}
g_mutex_free(pool->lock);
g_slice_free(liBackendPool_p, pool);
}
liBackendPool* li_backend_pool_new(const liBackendConfig *config) {
liBackendPool_p *pool = g_slice_new0(liBackendPool_p);
pool->public.config = config;
pool->lock = g_mutex_new();
pool->ts_disabled_till = 0;
return &pool->public;
}
void li_backend_pool_free(liBackendPool *bpool) {
liBackendPool_p *pool = LI_CONTAINER_OF(bpool, liBackendPool_p, public);
g_mutex_lock(pool->lock);
LI_FORCE_ASSERT(0 == pool->active);
LI_FORCE_ASSERT(!pool->shutdown);
pool->shutdown = TRUE;
g_mutex_unlock(pool->lock);
if (pool->worker_pools == NULL) {
backend_pool_worker_shutdown_done(NULL, pool, NULL, TRUE);
} else {
liServer *srv = pool->worker_pools[0].wrk->srv;
/*ERROR(srv, "li_backend_pool_free: %p, reserved: %i, idle: %i, pending: %i", (void*) pool,
pool->reserved, pool->idle, pool->pending); */
li_collect_start_global(srv, backend_pool_worker_shutdown, pool, backend_pool_worker_shutdown_done, NULL);
}
}
liBackendResult li_backend_get(liVRequest *vr, liBackendPool *bpool, liBackendConnection **pbcon, liBackendWait **pbwait) {
liBackendPool_p *pool = LI_CONTAINER_OF(bpool, liBackendPool_p, public);
liBackendWorkerPool *wpool;
liBackendResult result = LI_BACKEND_TIMEOUT;
liBackendWait *bwait = NULL;
LI_FORCE_ASSERT(pbcon);
LI_FORCE_ASSERT(pbwait);
g_mutex_lock(pool->lock);
S_backend_pool_init(vr->wrk, pool);
wpool = &pool->worker_pools[vr->wrk->ndx];
if (*pbwait) {
bwait = *pbwait;
LI_FORCE_ASSERT(vr == bwait->vr);
} else if (pool->ts_disabled_till > li_cur_ts(vr->wrk)) {
goto out;
} else {
if (wpool->idle > 0) {
/* shortcut without backend_pool_distribute */
liBackendConnection_p *con = g_ptr_array_index(wpool->connections, wpool->active + wpool->reserved);
con->active = TRUE;
S_backend_pool_worker_insert_con(pool, NULL, con);
*pbcon = &con->public;
result = LI_BACKEND_SUCCESS;
li_event_set_keep_loop_alive(&con->public.watcher, TRUE);
if (pool->public.config->watch_for_close) {
li_event_stop(&con->public.watcher);
li_event_set_callback(&con->public.watcher, NULL);
}
li_waitqueue_remove(&wpool->idle_queue, &con->timeout_elem);
goto out;
}
bwait = g_slice_new0(liBackendWait);
bwait->vr = vr;
bwait->vr_ref = li_vrequest_get_ref(vr);
bwait->ts_started = li_cur_ts(vr->wrk);
*pbwait = bwait;
if (pool->public.config->max_connections <= 0) {
g_queue_push_tail_link(&wpool->wait_queue, &bwait->wait_queue_link);
} else {
g_queue_push_tail_link(&pool->wait_queue, &bwait->wait_queue_link);
S_backend_pool_update_wait_queue_timer(wpool);
}
S_backend_pool_distribute(pool, vr->wrk);
}
LI_FORCE_ASSERT(bwait);
if (bwait->failed) {
bwait->vr = NULL;
li_job_ref_release(bwait->vr_ref);
bwait->vr_ref = NULL;
bwait->failed = FALSE;
g_slice_free(liBackendWait, bwait);
*pbwait = NULL;
result = LI_BACKEND_TIMEOUT;
goto out;
}
if (bwait->con && bwait->con->worker == vr->wrk) {
liBackendConnection_p *con = bwait->con;
bwait->con = NULL;
bwait->vr = NULL;
li_job_ref_release(bwait->vr_ref);
bwait->vr_ref = NULL;
g_slice_free(liBackendWait, bwait);
*pbwait = NULL;
*pbcon = &con->public;
con->wait = NULL;
con->active = TRUE;
S_backend_pool_worker_insert_con(pool, NULL, con);
result = LI_BACKEND_SUCCESS;
li_event_set_keep_loop_alive(&con->public.watcher, TRUE);
if (pool->public.config->watch_for_close) {
li_event_stop(&con->public.watcher);
li_event_set_callback(&con->public.watcher, NULL);
}
li_waitqueue_remove(&wpool->idle_queue, &con->timeout_elem);
goto out;
}
result = LI_BACKEND_WAIT;
out:
g_mutex_unlock(pool->lock);
return result;
}
void li_backend_wait_stop(liVRequest *vr, liBackendPool *bpool, liBackendWait **pbwait) {
liBackendPool_p *pool = LI_CONTAINER_OF(bpool, liBackendPool_p, public);
liBackendWait *bwait;
LI_FORCE_ASSERT(pbwait);
bwait = *pbwait;
if (!bwait) return;
*pbwait = NULL;
LI_FORCE_ASSERT(vr == bwait->vr);
if (bwait->failed) {
bwait->vr = NULL;
li_job_ref_release(bwait->vr_ref);
bwait->vr_ref = NULL;
bwait->failed = FALSE;
g_slice_free(liBackendWait, bwait);
return;
}
g_mutex_lock(pool->lock);
if (!bwait->failed) {
if (bwait->con) {
bwait->con->wait = NULL;
bwait->con->active = FALSE;
S_backend_pool_worker_insert_con(pool, NULL, bwait->con);
bwait->con = NULL;
} else if (pool->public.config->max_connections <= 0) {
liBackendWorkerPool *wpool = &pool->worker_pools[bwait->vr->wrk->ndx];
g_queue_unlink(&wpool->wait_queue, &bwait->wait_queue_link);
} else {
g_queue_unlink(&pool->wait_queue, &bwait->wait_queue_link);
}
}
bwait->vr = NULL;
li_job_ref_release(bwait->vr_ref);
bwait->vr_ref = NULL;
g_slice_free(liBackendWait, bwait);
g_mutex_unlock(pool->lock);
}
void li_backend_put(liWorker *wrk, liBackendPool *bpool, liBackendConnection *bcon, gboolean closecon) {
liBackendPool_p *pool = LI_CONTAINER_OF(bpool, liBackendPool_p, public);
liBackendConnection_p *con = LI_CONTAINER_OF(bcon, liBackendConnection_p, public);
liBackendWorkerPool *wpool = &pool->worker_pools[wrk->ndx];
++con->requests;
con->active = FALSE;
if (-1 == li_event_io_fd(&con->public.watcher) || closecon
|| (pool->public.config->max_requests > 0 && con->requests >= pool->public.config->max_requests)
|| (0 == pool->public.config->idle_timeout)) {
backend_connection_close(pool, con, FALSE);
} else {
g_mutex_lock(pool->lock);
pool->ts_disabled_till = 0;
li_event_set_keep_loop_alive(&con->public.watcher, FALSE);
if (pool->public.config->watch_for_close) {
li_event_set_callback(&con->public.watcher, backend_con_watch_for_close_cb);
li_event_io_set_events(&con->public.watcher, LI_EV_READ);
li_event_start(&con->public.watcher);
}
li_waitqueue_push(&wpool->idle_queue, &con->timeout_elem);
S_backend_pool_worker_insert_con(pool, NULL, con);
S_backend_pool_distribute(pool, wrk);
g_mutex_unlock(pool->lock);
}
}
void li_backend_connection_closed(liBackendPool *bpool, liBackendConnection *bcon) {
liBackendPool_p *pool = LI_CONTAINER_OF(bpool, liBackendPool_p, public);
liBackendConnection_p *con = LI_CONTAINER_OF(bcon, liBackendConnection_p, public);
backend_connection_close(pool, con, FALSE);
}