mirror of https://github.com/yandex/odyssey.git
259 lines
6.3 KiB
C
259 lines
6.3 KiB
C
|
|
/*
|
|
* Odyssey.
|
|
*
|
|
* Scalable PostgreSQL connection pooler.
|
|
*/
|
|
|
|
#include <stdlib.h>
|
|
#include <stdarg.h>
|
|
#include <stdint.h>
|
|
#include <stdio.h>
|
|
#include <string.h>
|
|
#include <inttypes.h>
|
|
#include <signal.h>
|
|
|
|
#include <machinarium.h>
|
|
#include <shapito.h>
|
|
|
|
#include "sources/macro.h"
|
|
#include "sources/version.h"
|
|
#include "sources/atomic.h"
|
|
#include "sources/util.h"
|
|
#include "sources/error.h"
|
|
#include "sources/list.h"
|
|
#include "sources/pid.h"
|
|
#include "sources/id.h"
|
|
#include "sources/logger.h"
|
|
#include "sources/daemon.h"
|
|
#include "sources/config.h"
|
|
#include "sources/config_reader.h"
|
|
#include "sources/msg.h"
|
|
#include "sources/global.h"
|
|
#include "sources/stat.h"
|
|
#include "sources/server.h"
|
|
#include "sources/server_pool.h"
|
|
#include "sources/client.h"
|
|
#include "sources/client_pool.h"
|
|
#include "sources/route_id.h"
|
|
#include "sources/route.h"
|
|
#include "sources/route_pool.h"
|
|
#include "sources/io.h"
|
|
#include "sources/instance.h"
|
|
#include "sources/router_cancel.h"
|
|
#include "sources/router.h"
|
|
#include "sources/system.h"
|
|
#include "sources/worker.h"
|
|
#include "sources/frontend.h"
|
|
#include "sources/backend.h"
|
|
#include "sources/cron.h"
|
|
|
|
static int
|
|
od_cron_stat_cb(od_route_t *route, od_stat_t *current, od_stat_t *avg,
|
|
void *arg)
|
|
{
|
|
od_router_t *router = arg;
|
|
od_instance_t *instance = router->global->instance;
|
|
|
|
/* update route stats */
|
|
route->stats_prev = *current;
|
|
|
|
if (! instance->config.log_stats)
|
|
return 0;
|
|
|
|
od_log(&instance->logger, "stats", NULL, NULL,
|
|
"[%.*s.%.*s] %d clients, "
|
|
"%d active servers, "
|
|
"%d idle servers, "
|
|
"%" PRIu64 " transactions/sec (%" PRIu64 " usec) "
|
|
"%" PRIu64 " queries/sec (%" PRIu64 " usec) "
|
|
"%" PRIu64 " in bytes/sec, "
|
|
"%" PRIu64 " out bytes/sec",
|
|
route->id.database_len - 1,
|
|
route->id.database,
|
|
route->id.user_len - 1,
|
|
route->id.user,
|
|
od_clientpool_total(&route->client_pool),
|
|
route->server_pool.count_active,
|
|
route->server_pool.count_idle,
|
|
avg->count_tx,
|
|
avg->tx_time,
|
|
avg->count_query,
|
|
avg->query_time,
|
|
avg->recv_client,
|
|
avg->recv_server);
|
|
|
|
return 0;
|
|
}
|
|
|
|
static inline void
|
|
od_cron_stat(od_router_t *router)
|
|
{
|
|
od_instance_t *instance = router->global->instance;
|
|
od_cron_t *cron = router->global->cron;
|
|
|
|
if (router->route_pool.count == 0)
|
|
return;
|
|
|
|
if (instance->config.log_stats)
|
|
{
|
|
int stream_count = 0;
|
|
int stream_count_allocated = 0;
|
|
int stream_total_allocated = 0;
|
|
int stream_cache_size = 0;
|
|
shapito_cache_stat(&instance->stream_cache, &stream_count,
|
|
&stream_count_allocated, &stream_total_allocated,
|
|
&stream_cache_size);
|
|
int count_machine = 0;
|
|
int count_coroutine = 0;
|
|
int count_coroutine_cache = 0;
|
|
machinarium_stat(&count_machine, &count_coroutine,
|
|
&count_coroutine_cache);
|
|
od_log(&instance->logger, "stats", NULL, NULL,
|
|
"clients %d, stream cache (%d:%d allocated, %d cached %d bytes), "
|
|
"coroutines (%d active, %d cached)",
|
|
router->clients,
|
|
stream_count_allocated,
|
|
stream_total_allocated,
|
|
stream_count,
|
|
stream_cache_size,
|
|
count_coroutine,
|
|
count_coroutine_cache);
|
|
}
|
|
|
|
/* update stats per route */
|
|
od_routepool_stat(&router->route_pool, od_cron_stat_cb,
|
|
cron->stat_time_us,
|
|
router);
|
|
|
|
/* update current stat time mark */
|
|
cron->stat_time_us = machine_time();
|
|
}
|
|
|
|
static inline int
|
|
od_cron_expire_mark(od_server_t *server, void *arg)
|
|
{
|
|
od_router_t *router = arg;
|
|
od_instance_t *instance = router->global->instance;
|
|
od_route_t *route = server->route;
|
|
|
|
/* expire by time-to-live */
|
|
if (! route->config->pool_ttl)
|
|
return 0;
|
|
|
|
od_debug(&instance->logger, "expire", NULL, server,
|
|
"idle time: %d",
|
|
server->idle_time);
|
|
|
|
if (server->idle_time < route->config->pool_ttl) {
|
|
server->idle_time++;
|
|
return 0;
|
|
}
|
|
od_serverpool_set(&route->server_pool, server,
|
|
OD_SEXPIRE);
|
|
return 0;
|
|
}
|
|
|
|
static inline void
|
|
od_cron_expire(od_cron_t *cron)
|
|
{
|
|
od_router_t *router = cron->global->router;
|
|
od_instance_t *instance = cron->global->instance;
|
|
|
|
/* Idle servers expire.
|
|
*
|
|
* It is important that mark logic stage must not yield
|
|
* to maintain iterator consistency.
|
|
*
|
|
* mark:
|
|
*
|
|
* - If a server idle time is equal to ttl, then move
|
|
* it to the EXPIRE queue.
|
|
*
|
|
* - If a server config marked as obsolete and route has
|
|
* no remaining clients, then move it to the EXPIRE queue.
|
|
*
|
|
* - Add plus one idle second on each traversal.
|
|
*
|
|
* sweep:
|
|
*
|
|
* - Foreach servers in EXPIRE queue, send Terminate
|
|
* and close the connection.
|
|
*/
|
|
|
|
/* mark */
|
|
od_routepool_server_foreach(&router->route_pool, OD_SIDLE,
|
|
od_cron_expire_mark,
|
|
router);
|
|
|
|
/* sweep */
|
|
for (;;) {
|
|
od_server_t *server;
|
|
server = od_routepool_next(&router->route_pool, OD_SEXPIRE);
|
|
if (server == NULL)
|
|
break;
|
|
od_debug(&instance->logger, "expire", NULL, server,
|
|
"closing idle server connection (%d secs)",
|
|
server->idle_time);
|
|
server->idle_time = 0;
|
|
|
|
od_route_t *route = server->route;
|
|
server->route = NULL;
|
|
od_serverpool_set(&route->server_pool, server, OD_SUNDEF);
|
|
|
|
if (instance->is_shared)
|
|
machine_io_attach(server->io);
|
|
|
|
od_backend_close_connection(server);
|
|
od_backend_close(server);
|
|
}
|
|
|
|
/* cleanup unused dynamic routes */
|
|
od_routepool_gc(&router->route_pool);
|
|
}
|
|
|
|
static void
|
|
od_cron(void *arg)
|
|
{
|
|
od_cron_t *cron = arg;
|
|
od_router_t *router = cron->global->router;
|
|
od_instance_t *instance = cron->global->instance;
|
|
|
|
cron->stat_time_us = machine_time();
|
|
|
|
int stats_tick = 0;
|
|
for (;;)
|
|
{
|
|
/* mark and sweep expired idle server connections */
|
|
od_cron_expire(cron);
|
|
|
|
/* update statistics */
|
|
if (++stats_tick >= instance->config.stats_interval) {
|
|
od_cron_stat(router);
|
|
stats_tick = 0;
|
|
}
|
|
|
|
/* 1 second soft interval */
|
|
machine_sleep(1000);
|
|
}
|
|
}
|
|
|
|
void od_cron_init(od_cron_t *cron, od_global_t *global)
|
|
{
|
|
cron->global = global;
|
|
cron->stat_time_us = 0;
|
|
}
|
|
|
|
int od_cron_start(od_cron_t *cron)
|
|
{
|
|
od_instance_t *instance = cron->global->instance;
|
|
int64_t coroutine_id;
|
|
coroutine_id = machine_coroutine_create(od_cron, cron);
|
|
if (coroutine_id == -1) {
|
|
od_error(&instance->logger, "cron", NULL, NULL,
|
|
"failed to start cron coroutine");
|
|
return -1;
|
|
}
|
|
return 0;
|
|
}
|