Rooba
/
odyssey
mirror of https://github.com/yandex/odyssey.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

odyssey: remake system_t as global_t

This commit is contained in:
Dmitry Simonenko 2018-03-13 16:17:27 +03:00
parent 5862c1762f
commit 20a092098e
30 changed files with 153 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
sources/auth.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -52,7 +52,7 @@
static inline int
od_auth_frontend_cleartext(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
/* AuthenticationCleartextPassword */
shapito_stream_t *stream = client->stream;
@ -105,7 +105,7 @@ od_auth_frontend_cleartext(od_client_t *client)
shapito_password_init(&client_password);
if (client->config->auth_query) {
rc = od_auth_query(client->system,
rc = od_auth_query(client->global,
stream,
client->config,
client->startup.user,
@ -144,7 +144,7 @@ od_auth_frontend_cleartext(od_client_t *client)
static inline int
od_auth_frontend_md5(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
/* generate salt */
uint32_t salt = shapito_password_salt(&client->key);
@ -204,7 +204,7 @@ od_auth_frontend_md5(od_client_t *client)
shapito_password_init(&query_password);
if (client->config->auth_query) {
rc = od_auth_query(client->system,
rc = od_auth_query(client->global,
stream,
client->config,
client->startup.user,
@ -262,7 +262,7 @@ od_auth_frontend_md5(od_client_t *client)
static inline int
od_auth_frontend_block(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
od_log(&instance->logger, "auth", client, NULL,
"user '%s.%s' is blocked",
shapito_parameter_value(client->startup.database),
@ -274,7 +274,7 @@ od_auth_frontend_block(od_client_t *client)
int od_auth_frontend(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
/* authentication mode */
int rc;
@ -318,7 +318,7 @@ int od_auth_frontend(od_client_t *client)
static inline int
od_auth_backend_cleartext(od_server_t *server, shapito_stream_t *stream)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
od_route_t *route = server->route;
assert(route != NULL);
@ -366,7 +366,7 @@ static inline int
od_auth_backend_md5(od_server_t *server, shapito_stream_t *stream,
char salt[4])
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
od_route_t *route = server->route;
assert(route != NULL);
@ -439,7 +439,8 @@ od_auth_backend_md5(od_server_t *server, shapito_stream_t *stream,
int od_auth_backend(od_server_t *server, shapito_stream_t *stream)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
assert(*stream->start == SHAPITO_BE_AUTHENTICATION);
uint32_t auth_type;

10
sources/auth_query.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -55,7 +55,7 @@ od_auth_query_do(od_server_t *server, shapito_stream_t *stream,
char *query, int len,
shapito_password_t *result)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
od_debug(&instance->logger, "auth_query", server->client, server,
"%s", query);
@ -212,20 +212,20 @@ od_auth_query_format(od_configroute_t *config, shapito_parameter_t *user,
return dst_pos - output;
}
int od_auth_query(od_system_t *system,
int od_auth_query(od_global_t *global,
shapito_stream_t *stream,
od_configroute_t *config,
shapito_parameter_t *user,
shapito_password_t *password)
{
od_instance_t *instance = system->instance;
od_instance_t *instance = global->instance;
/* create internal auth client */
od_client_t *auth_client;
auth_client = od_client_allocate();
if (auth_client == NULL)
return -1;
auth_client->system = system;
auth_client->global = global;
od_idmgr_generate(&instance->id_mgr, &auth_client->id, "a");

2
sources/auth_query.h
View File

@ -7,7 +7,7 @@
* Advanced PostgreSQL connection pooler.
*/
int od_auth_query(od_system_t*, shapito_stream_t*, od_configroute_t*,
int od_auth_query(od_global_t*, shapito_stream_t*, od_configroute_t*,
shapito_parameter_t*, shapito_password_t*);
#endif /* OD_AUTH_QUERY_H */

24
sources/backend.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -78,7 +78,7 @@ od_backend_terminate(od_server_t *server, shapito_stream_t *stream)
void od_backend_close_connection(od_server_t *server)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
if (server->io == NULL)
return;
@ -104,7 +104,7 @@ void od_backend_close_connection(od_server_t *server)
void od_backend_error(od_server_t *server, char *context, char *data, int size)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
shapito_fe_error_t error;
int rc;
rc = shapito_fe_read_error(&error, data, size);
@ -132,7 +132,7 @@ void od_backend_error(od_server_t *server, char *context, char *data, int size)
int od_backend_ready(od_server_t *server, char *context,
char *data, int size)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
int status;
int rc;
rc = shapito_fe_read_ready(&status, data, size);
@ -160,7 +160,7 @@ int od_backend_ready(od_server_t *server, char *context,
static inline int
od_backend_startup(od_server_t *server, shapito_stream_t *stream)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
od_route_t *route = server->route;
shapito_stream_reset(stream);
@ -259,7 +259,7 @@ od_backend_connect_to(od_server_t *server,
od_configstorage_t *server_config,
char *context)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
assert(server->io == NULL);
/* create io handle */
@ -324,7 +324,7 @@ od_backend_connect_to(od_server_t *server,
int od_backend_connect(od_server_t *server, shapito_stream_t *stream,
char *context)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
od_route_t *route = server->route;
assert(route != NULL);
@ -355,7 +355,7 @@ int od_backend_connect_cancel(od_server_t *server,
od_configstorage_t *server_config,
shapito_key_t *key)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
/* connect to server */
int rc;
rc = od_backend_connect_to(server, stream, server_config, "cancel");
@ -379,7 +379,7 @@ int od_backend_ready_wait(od_server_t *server, shapito_stream_t *stream,
char *context, int count,
uint32_t time_ms)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
int ready = 0;
for (;;)
@ -443,7 +443,7 @@ int od_backend_query(od_server_t *server, shapito_stream_t *stream,
char *context,
char *query, int len)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
shapito_stream_reset(stream);
int rc;
rc = shapito_fe_write_query(stream, query, len);
@ -468,7 +468,7 @@ int
od_backend_deploy(od_server_t *server, char *context,
char *request, int request_size)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
int rc;
switch (*request) {
case SHAPITO_BE_PARAMETER_STATUS:
@ -510,7 +510,7 @@ od_backend_deploy(od_server_t *server, char *context,
int od_backend_deploy_wait(od_server_t *server, shapito_stream_t *stream,
char *context, uint32_t time_ms)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
while (server->deploy_sync > 0) {
shapito_stream_reset(stream);
int rc;

8
sources/cancel.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -50,20 +50,20 @@
#include "sources/tls.h"
#include "sources/cancel.h"
int od_cancel(od_system_t *system,
int od_cancel(od_global_t *global,
shapito_stream_t *stream,
od_configstorage_t *server_config,
shapito_key_t *key,
od_id_t *server_id)
{
od_instance_t *instance = system->instance;
od_instance_t *instance = global->instance;
od_log(&instance->logger, "cancel", NULL, NULL,
"cancel for %s%.*s",
server_id->id_prefix,
sizeof(server_id->id), server_id->id);
od_server_t server;
od_server_init(&server);
server.system = system;
server.global = global;
od_backend_connect_cancel(&server, stream, server_config, key);
od_backend_close_connection(&server);
od_backend_close(&server);

2
sources/cancel.h
View File

@ -7,7 +7,7 @@
* Advanced PostgreSQL connection pooler.
*/
int od_cancel(od_system_t*, shapito_stream_t*, od_configstorage_t*, shapito_key_t*, od_id_t*);
int od_cancel(od_global_t*, shapito_stream_t*, od_configstorage_t*, shapito_key_t*, od_id_t*);
int od_cancel_find(od_routepool_t*, shapito_key_t*, od_routercancel_t*);
#endif

4
sources/client.h
View File

@ -35,7 +35,7 @@ struct od_client
shapito_stream_t *stream;
od_server_t *server;
void *route;
od_system_t *system;
od_global_t *global;
od_list_t link_pool;
od_list_t link;
};
@ -52,7 +52,7 @@ od_client_init(od_client_t *client)
client->config_listen = NULL;
client->server = NULL;
client->route = NULL;
client->system = NULL;
client->global = NULL;
client->time_accept = 0;
client->time_setup = 0;
client->stream = NULL;

2
sources/client_pool.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"

24
sources/console.c
View File

@ -31,7 +31,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -153,7 +153,7 @@ od_console_show_stats_callback(char *database,
static inline int
od_console_show_stats(od_client_t *client)
{
od_router_t *router = client->system->router;
od_router_t *router = client->global->router;
shapito_stream_t *stream = client->stream;
shapito_stream_reset(stream);
int rc;
@ -285,7 +285,7 @@ od_console_show_servers_callback(od_server_t *server, void *arg)
static inline int
od_console_show_servers(od_client_t *client)
{
od_router_t *router = client->system->router;
od_router_t *router = client->global->router;
shapito_stream_t *stream = client->stream;
shapito_stream_reset(stream);
int rc;
@ -432,7 +432,7 @@ od_console_show_clients_callback(od_client_t *client, void *arg)
static inline int
od_console_show_clients(od_client_t *client)
{
od_router_t *router = client->system->router;
od_router_t *router = client->global->router;
shapito_stream_t *stream = client->stream;
shapito_stream_reset(stream);
int rc;
@ -507,7 +507,7 @@ od_console_show_lists_callback(od_server_t *server, void *arg)
static inline int
od_console_show_lists(od_client_t *client)
{
od_router_t *router = client->system->router;
od_router_t *router = client->global->router;
shapito_stream_t *stream = client->stream;
shapito_stream_reset(stream);
@ -631,7 +631,7 @@ od_console_query_set(od_client_t *client, od_parser_t *parser)
static inline int
od_console_query(od_console_t *console, od_msgconsole_t *msg_console)
{
od_instance_t *instance = console->system->instance;
od_instance_t *instance = console->global->instance;
od_client_t *client = msg_console->client;
int rc;
@ -702,7 +702,7 @@ static void
od_console(void *arg)
{
od_console_t *console = arg;
od_instance_t *instance = console->system->instance;
od_instance_t *instance = console->global->instance;
(void)instance;
for (;;) {
@ -734,15 +734,15 @@ od_console(void *arg)
}
}
void od_console_init(od_console_t *console, od_system_t *system)
void od_console_init(od_console_t *console, od_global_t *global)
{
console->system = system;
console->global = global;
console->channel = NULL;
}
int od_console_start(od_console_t *console)
{
od_instance_t *instance = console->system->instance;
od_instance_t *instance = console->global->instance;
console->channel = machine_channel_create(instance->is_shared);
if (console->channel == NULL) {
@ -764,8 +764,8 @@ static od_consolestatus_t
od_console_do(od_client_t *client, od_msg_t msg_type, char *request, int request_len,
int wait_for_response)
{
od_console_t *console = client->system->console;
od_instance_t *instance = console->system->instance;
od_console_t *console = client->global->console;
od_instance_t *instance = console->global->instance;
/* send request to console */
machine_msg_t *msg;

4
sources/console.h
View File

@ -16,11 +16,11 @@ typedef enum
} od_consolestatus_t;
struct od_console_t {
od_system_t *system;
machine_channel_t *channel;
od_global_t *global;
};
void od_console_init(od_console_t*, od_system_t*);
void od_console_init(od_console_t*, od_global_t*);
int od_console_start(od_console_t*);
od_consolestatus_t

20
sources/cron.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -62,7 +62,7 @@ od_cron_stats_server(od_server_t *server, void *arg)
static inline void
od_cron_stats(od_router_t *router)
{
od_instance_t *instance = router->system->instance;
od_instance_t *instance = router->global->instance;
if (router->route_pool.count == 0)
return;
@ -206,7 +206,7 @@ static inline int
od_cron_expire_mark(od_server_t *server, void *arg)
{
od_router_t *router = arg;
od_instance_t *instance = router->system->instance;
od_instance_t *instance = router->global->instance;
od_route_t *route = server->route;
/* expire by server config obsoletion */
@ -238,8 +238,8 @@ od_cron_expire_mark(od_server_t *server, void *arg)
static inline void
od_cron_expire(od_cron_t *cron)
{
od_router_t *router = cron->system->router;
od_instance_t *instance = cron->system->instance;
od_router_t *router = cron->global->router;
od_instance_t *instance = cron->global->instance;
/* Idle servers expire.
*
@ -298,8 +298,8 @@ static void
od_cron(void *arg)
{
od_cron_t *cron = arg;
od_router_t *router = cron->system->router;
od_instance_t *instance = cron->system->instance;
od_router_t *router = cron->global->router;
od_instance_t *instance = cron->global->instance;
int stats_tick = 0;
for (;;)
@ -318,14 +318,14 @@ od_cron(void *arg)
}
}
void od_cron_init(od_cron_t *cron, od_system_t *system)
void od_cron_init(od_cron_t *cron, od_global_t *global)
{
cron->system = system;
cron->global = global;
}
int od_cron_start(od_cron_t *cron)
{
od_instance_t *instance = cron->system->instance;
od_instance_t *instance = cron->global->instance;
int64_t coroutine_id;
coroutine_id = machine_coroutine_create(od_cron, cron);
if (coroutine_id == -1) {

4
sources/cron.h
View File

@ -11,10 +11,10 @@ typedef struct od_cron od_cron_t;
struct od_cron
{
od_system_t *system;
od_global_t *global;
};
void od_cron_init(od_cron_t*, od_system_t*);
void od_cron_init(od_cron_t*, od_global_t*);
int od_cron_start(od_cron_t*);
#endif /* OD_CRON_H */

4
sources/deploy.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -77,7 +77,7 @@ int od_deploy_write(od_server_t *server, char *context,
shapito_stream_t *stream,
shapito_parameters_t *params)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
int rc;
/* discard */

30
sources/frontend.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -69,7 +69,7 @@ typedef enum {
void od_frontend_close(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
assert(client->route == NULL);
assert(client->server == NULL);
if (client->stream) {
@ -86,7 +86,7 @@ void od_frontend_close(od_client_t *client)
static inline int
od_frontend_error_fwd(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
od_server_t *server = client->server;
assert(server != NULL);
assert(server->stats.count_error != 0);
@ -171,7 +171,7 @@ int od_frontend_error(od_client_t *client, char *code, char *fmt, ...)
static int
od_frontend_startup_read(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
shapito_stream_t *stream = client->stream;
shapito_stream_reset(stream);
@ -206,7 +206,7 @@ od_frontend_startup_read(od_client_t *client)
static int
od_frontend_startup(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
int rc;
rc = od_frontend_startup_read(client);
@ -266,7 +266,7 @@ od_frontend_key(od_client_t *client)
static inline od_frontend_rc_t
od_frontend_attach(od_client_t *client, char *context)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
od_routerstatus_t status;
od_server_t *server;
@ -357,7 +357,7 @@ od_frontend_setup_console(shapito_stream_t *stream)
static inline od_frontend_rc_t
od_frontend_setup(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
od_route_t *route = client->route;
shapito_stream_t *stream = client->stream;
shapito_stream_reset(stream);
@ -459,14 +459,14 @@ od_frontend_setup(od_client_t *client)
static inline int
od_frontend_stream_hit_limit(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
return shapito_stream_used(client->stream) >= instance->config.pipeline;
}
static od_frontend_rc_t
od_frontend_local(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
int rc;
shapito_stream_t *stream = client->stream;
@ -518,7 +518,7 @@ od_frontend_local(od_client_t *client)
static inline od_frontend_rc_t
od_frontend_remote_client(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
od_server_t *server = client->server;
shapito_stream_t *stream = client->stream;
shapito_stream_reset(stream);
@ -625,7 +625,7 @@ od_frontend_remote_client(od_client_t *client)
static inline od_frontend_rc_t
od_frontend_remote_server(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
od_route_t *route = client->route;
od_server_t *server = client->server;
shapito_stream_t *stream = client->stream;
@ -745,7 +745,7 @@ od_frontend_remote_server(od_client_t *client)
static od_frontend_rc_t
od_frontend_remote(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
assert(client->stream != NULL);
machine_io_t *io_ready[2];
@ -797,7 +797,7 @@ static void
od_frontend_cleanup(od_client_t *client, char *context,
od_frontend_rc_t status)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
int rc;
od_server_t *server = client->server;
@ -924,7 +924,7 @@ od_frontend_cleanup(od_client_t *client, char *context,
void od_frontend(void *arg)
{
od_client_t *client = arg;
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
/* log client connection */
if (instance->config.log_session) {
@ -964,7 +964,7 @@ void od_frontend(void *arg)
od_routercancel_init(&cancel);
rc = od_router_cancel(client, &cancel);
if (rc == 0) {
od_cancel(client->system, client->stream, cancel.config,
od_cancel(client->global, client->stream, cancel.config,
&cancel.key, &cancel.id);
od_routercancel_free(&cancel);
}

10
sources/system.h → sources/global.h
View File

@ -1,5 +1,5 @@
#ifndef OD_SYSTEM_H
#define OD_SYSTEM_H
#ifndef OD_GLOBAL_H
#define OD_GLOBAL_H
/*
* Odyssey.
@ -7,9 +7,9 @@
* Advanced PostgreSQL connection pooler.
*/
typedef struct od_system od_system_t;
typedef struct od_global od_global_t;
struct od_system
struct od_global
{
void *instance;
void *pooler;
@ -19,4 +19,4 @@ struct od_system
void *worker_pool;
};
#endif /* OD_SYSTEM_H */
#endif /* OD_GLOBAL_H */

26
sources/instance.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -191,7 +191,7 @@ int od_instance_main(od_instance_t *instance, int argc, char **argv)
/* is multi-worker deploy */
instance->is_shared = instance->config.workers > 1;
/* prepare system services */
/* prepare global services */
od_pooler_t pooler;
od_pooler_init(&pooler, instance);
@ -200,18 +200,18 @@ int od_instance_main(od_instance_t *instance, int argc, char **argv)
od_cron_t cron;
od_workerpool_t worker_pool;
od_system_t *system;
system = &pooler.system;
system->instance = instance;
system->pooler = &pooler;
system->router = &router;
system->console = &console;
system->cron = &cron;
system->worker_pool = &worker_pool;
od_global_t *global;
global = &pooler.global;
global->instance = instance;
global->pooler = &pooler;
global->router = &router;
global->console = &console;
global->cron = &cron;
global->worker_pool = &worker_pool;
od_router_init(&router, system);
od_console_init(&console, system);
od_cron_init(&cron, system);
od_router_init(&router, global);
od_console_init(&console, global);
od_cron_init(&cron, global);
od_workerpool_init(&worker_pool);
/* start pooler machine thread */

2
sources/logger.c
View File

@ -36,7 +36,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"

30
sources/pooler.c
View File

@ -32,7 +32,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -55,7 +55,7 @@ static inline void
od_pooler_server(void *arg)
{
od_poolerserver_t *server = arg;
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
for (;;)
{
@ -108,7 +108,7 @@ od_pooler_server(void *arg)
msg = machine_msg_create(OD_MCLIENT_NEW, sizeof(od_client_t*));
char *msg_data = machine_msg_get_data(msg);
memcpy(msg_data, &client, sizeof(od_client_t*));
od_workerpool_t *worker_pool = server->system->worker_pool;
od_workerpool_t *worker_pool = server->global->worker_pool;
od_workerpool_feed(worker_pool, msg);
}
}
@ -117,7 +117,7 @@ static inline int
od_pooler_server_start(od_pooler_t *pooler, od_configlisten_t *config,
struct addrinfo *addr)
{
od_instance_t *instance = pooler->system.instance;
od_instance_t *instance = pooler->global.instance;
od_poolerserver_t *server;
server = malloc(sizeof(od_poolerserver_t));
if (server == NULL) {
@ -127,7 +127,7 @@ od_pooler_server_start(od_pooler_t *pooler, od_configlisten_t *config,
}
server->config = config;
server->addr = addr;
server->system = &pooler->system;
server->global = &pooler->global;
/* create server tls */
if (server->config->tls_mode != OD_TLS_DISABLE) {
@ -191,7 +191,7 @@ od_pooler_server_start(od_pooler_t *pooler, od_configlisten_t *config,
static inline int
od_pooler_main(od_pooler_t *pooler)
{
od_instance_t *instance = pooler->system.instance;
od_instance_t *instance = pooler->global.instance;
int binded = 0;
od_list_t *i;
od_list_foreach(&instance->config.listen, i)
@ -249,7 +249,7 @@ od_pooler_main(od_pooler_t *pooler)
static inline void
od_pooler_config_import(od_pooler_t *pooler)
{
od_instance_t *instance = pooler->system.instance;
od_instance_t *instance = pooler->global.instance;
od_log(&instance->logger, "config", NULL, NULL, "importing changes from '%s'",
instance->config_file);
@ -295,7 +295,7 @@ static inline void
od_pooler_signal_handler(void *arg)
{
od_pooler_t *pooler = arg;
od_instance_t *instance = pooler->system.instance;
od_instance_t *instance = pooler->global.instance;
sigset_t mask;
sigemptyset(&mask);
@ -342,26 +342,26 @@ od_pooler(void *arg)
/* start router coroutine */
int rc;
od_router_t *router = pooler->system.router;
od_router_t *router = pooler->global.router;
rc = od_router_start(router);
if (rc == -1)
return;
/* start console coroutine */
od_console_t *console = pooler->system.console;
od_console_t *console = pooler->global.console;
rc = od_console_start(console);
if (rc == -1)
return;
/* start cron coroutine */
od_cron_t *cron = pooler->system.cron;
od_cron_t *cron = pooler->global.cron;
rc = od_cron_start(cron);
if (rc == -1)
return;
/* start worker threads */
od_workerpool_t *worker_pool = pooler->system.worker_pool;
rc = od_workerpool_start(worker_pool, &pooler->system, instance->config.workers);
od_workerpool_t *worker_pool = pooler->global.worker_pool;
rc = od_workerpool_start(worker_pool, &pooler->global, instance->config.workers);
if (rc == -1)
return;
@ -388,13 +388,13 @@ int od_pooler_init(od_pooler_t *pooler, od_instance_t *instance)
pooler->machine = -1;
pooler->instance = instance;
pooler->addr = NULL;
memset(&pooler->system, 0, sizeof(pooler->system));
memset(&pooler->global, 0, sizeof(pooler->global));
return 0;
}
int od_pooler_start(od_pooler_t *pooler)
{
od_instance_t *instance = pooler->system.instance;
od_instance_t *instance = pooler->global.instance;
pooler->machine = machine_create("pooler", od_pooler, pooler);
if (pooler->machine == -1) {
od_error(&instance->logger, "pooler", NULL, NULL,

4
sources/pooler.h
View File

@ -16,14 +16,14 @@ struct od_poolerserver
od_configlisten_t *config;
machine_io_t *io;
machine_tls_t *tls;
od_system_t *system;
od_global_t *global;
};
struct od_pooler
{
int64_t machine;
struct addrinfo *addr;
od_system_t system;
od_global_t global;
od_instance_t *instance;
};

6
sources/reset.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -53,7 +53,7 @@
int od_reset(od_server_t *server, shapito_stream_t *stream)
{
od_instance_t *instance = server->system->instance;
od_instance_t *instance = server->global->instance;
od_route_t *route = server->route;
/* server left in copy mode */
@ -127,7 +127,7 @@ int od_reset(od_server_t *server, shapito_stream_t *stream)
"not responded, cancel (#%d)",
wait_try_cancel);
wait_try_cancel++;
rc = od_cancel(server->system,
rc = od_cancel(server->global,
stream,
route->config->storage, &server->key,
&server->id);

2
sources/route_pool.c
View File

@ -27,7 +27,7 @@
#include "sources/config.h"
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"

30
sources/router.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -60,7 +60,7 @@ typedef struct
static od_route_t*
od_router_fwd(od_router_t *router, shapito_be_startup_t *startup)
{
od_instance_t *instance = router->system->instance;
od_instance_t *instance = router->global->instance;
assert(startup->database != NULL);
assert(startup->user != NULL);
@ -119,10 +119,10 @@ od_router_attacher(void *arg)
client = msg_attach->client;
od_instance_t *instance;
instance = client->system->instance;
instance = client->global->instance;
od_router_t *router;
router = client->system->router;
router = client->global->router;
od_route_t *route;
route = client->route;
@ -191,7 +191,7 @@ od_router_attacher(void *arg)
return;
}
od_idmgr_generate(&instance->id_mgr, &server->id, "s");
server->system = router->system;
server->global = router->global;
server->route = route;
on_attach:
@ -210,7 +210,7 @@ static inline void
od_router_wakeup(od_router_t *router, od_route_t *route)
{
od_instance_t *instance;
instance = router->system->instance;
instance = router->global->instance;
/* wake up first client waiting for route
* server connection */
if (route->client_pool.count_queue > 0) {
@ -230,7 +230,7 @@ static inline void
od_router(void *arg)
{
od_router_t *router = arg;
od_instance_t *instance = router->system->instance;
od_instance_t *instance = router->global->instance;
for (;;)
{
@ -466,17 +466,17 @@ od_router(void *arg)
}
}
void od_router_init(od_router_t *router, od_system_t *system)
void od_router_init(od_router_t *router, od_global_t *global)
{
od_routepool_init(&router->route_pool);
router->system = system;
router->global = global;
router->clients = 0;
router->channel = NULL;
}
int od_router_start(od_router_t *router)
{
od_instance_t *instance = router->system->instance;
od_instance_t *instance = router->global->instance;
router->channel = machine_channel_create(instance->is_shared);
if (router->channel == NULL) {
@ -497,8 +497,8 @@ int od_router_start(od_router_t *router)
static od_routerstatus_t
od_router_do(od_client_t *client, od_msg_t msg_type, od_routercancel_t *cancel)
{
od_router_t *router = client->system->router;
od_instance_t *instance = router->system->instance;
od_router_t *router = client->global->router;
od_instance_t *instance = router->global->instance;
/* send request to router */
machine_msg_t *msg;
@ -553,7 +553,7 @@ od_unroute(od_client_t *client)
od_routerstatus_t
od_router_attach(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
od_routerstatus_t status;
status = od_router_do(client, OD_MROUTER_ATTACH, NULL);
/* attach server io to clients machine context */
@ -568,7 +568,7 @@ od_router_attach(od_client_t *client)
od_routerstatus_t
od_router_detach(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
od_server_t *server = client->server;
if (instance->is_shared)
machine_io_detach(server->io);
@ -578,7 +578,7 @@ od_router_detach(od_client_t *client)
od_routerstatus_t
od_router_detach_and_unroute(od_client_t *client)
{
od_instance_t *instance = client->system->instance;
od_instance_t *instance = client->global->instance;
od_server_t *server = client->server;
if (instance->is_shared)
machine_io_detach(server->io);

4
sources/router.h
View File

@ -23,10 +23,10 @@ struct od_router
od_routepool_t route_pool;
machine_channel_t *channel;
int clients;
od_system_t *system;
od_global_t *global;
};
void od_router_init(od_router_t*, od_system_t*);
void od_router_init(od_router_t*, od_global_t*);
int od_router_start(od_router_t*);
od_routerstatus_t

4
sources/server.h
View File

@ -47,7 +47,7 @@ struct od_server
od_id_t last_client_id;
void *client;
void *route;
od_system_t *system;
od_global_t *global;
od_list_t link;
};
@ -57,7 +57,7 @@ od_server_init(od_server_t *server)
server->state = OD_SUNDEF;
server->route = NULL;
server->client = NULL;
server->system = NULL;
server->global = NULL;
server->io = NULL;
server->tls = NULL;
server->idle_time = 0;

2
sources/server_pool.c
View File

@ -27,7 +27,7 @@
#include "sources/config.h"
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"

2
sources/tls.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"

12
sources/worker.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -49,7 +49,7 @@ static inline void
od_worker(void *arg)
{
od_worker_t *worker = arg;
od_instance_t *instance = worker->system->instance;
od_instance_t *instance = worker->global->instance;
for (;;)
{
@ -65,7 +65,7 @@ od_worker(void *arg)
{
od_client_t *client;
client = *(od_client_t**)machine_msg_get_data(msg);
client->system = worker->system;
client->global = worker->global;
int64_t coroutine_id;
coroutine_id = machine_coroutine_create(od_frontend, client);
if (coroutine_id == -1) {
@ -89,16 +89,16 @@ od_worker(void *arg)
od_log(&instance->logger, "worker", NULL, NULL, "stopped");
}
void od_worker_init(od_worker_t *worker, od_system_t *system, int id)
void od_worker_init(od_worker_t *worker, od_global_t *global, int id)
{
worker->machine = -1;
worker->id = id;
worker->system = system;
worker->global = global;
}
int od_worker_start(od_worker_t *worker)
{
od_instance_t *instance = worker->system->instance;
od_instance_t *instance = worker->global->instance;
worker->task_channel = machine_channel_create(instance->is_shared);
if (worker->task_channel == NULL) {

4
sources/worker.h
View File

@ -14,10 +14,10 @@ struct od_worker
int64_t machine;
int id;
machine_channel_t *task_channel;
od_system_t *system;
od_global_t *global;
};
void od_worker_init(od_worker_t*, od_system_t*, int);
void od_worker_init(od_worker_t*, od_global_t*, int);
int od_worker_start(od_worker_t*);
#endif /* OD_WORKER_H */

6
sources/worker_pool.c
View File

@ -30,7 +30,7 @@
#include "sources/config_mgr.h"
#include "sources/config_reader.h"
#include "sources/msg.h"
#include "sources/system.h"
#include "sources/global.h"
#include "sources/server.h"
#include "sources/server_pool.h"
#include "sources/client.h"
@ -53,7 +53,7 @@ void od_workerpool_init(od_workerpool_t *pool)
pool->pool = NULL;
}
int od_workerpool_start(od_workerpool_t *pool, od_system_t *system, int count)
int od_workerpool_start(od_workerpool_t *pool, od_global_t *global, int count)
{
pool->pool = malloc(sizeof(od_worker_t) * count);
if (pool->pool == NULL)
@ -62,7 +62,7 @@ int od_workerpool_start(od_workerpool_t *pool, od_system_t *system, int count)
int i;
for (i = 0; i < count; i++) {
od_worker_t *worker = &pool->pool[i];
od_worker_init(worker, system, i);
od_worker_init(worker, global, i);
int rc;
rc = od_worker_start(worker);
if (rc == -1)

2
sources/worker_pool.h
View File

@ -17,7 +17,7 @@ struct od_workerpool
};
void od_workerpool_init(od_workerpool_t*);
int od_workerpool_start(od_workerpool_t*, od_system_t*, int);
int od_workerpool_start(od_workerpool_t*, od_global_t*, int);
void od_workerpool_feed(od_workerpool_t*, machine_msg_t*);
#endif /* OD_WORKER_POOL_H */