pgpool2.git / summary / log / commit / refs
commit 4ddcffc6341eedb6f30c22ff64fe3a67252be0ac
Author: Muhammad Usama <muhammad.usama@enterprisedb.com>
Commit: Muhammad Usama <m.usama@gmail.com>
Date: Thu Nov 07 13:25:29 2013 +0000
Integrating the exception and memory manager in pcp child process
src/main/pgpool_main.c | 2 +-
src/pcp_con/pcp_child.c | 1763 ++++++++++++++++++------------------
src/utils/pool_process_reporting.c | 10 +-
3 files changed, 879 insertions(+), 896 deletions(-)
diff --git a/src/main/pgpool_main.c b/src/main/pgpool_main.c
index 8adb4e73d..576a4c3ac 100644
--- a/src/main/pgpool_main.c
+++ b/src/main/pgpool_main.c
@@ -1711,7 +1711,7 @@ pool_get_process_list(int *array_size)
int i;
*array_size = pool_config->num_init_children;
- array = calloc(*array_size, sizeof(int));
+ array = palloc0(*array_size * sizeof(int));
for (i = 0; i < *array_size; i++)
array[i] = process_info[i].pid;
diff --git a/src/pcp_con/pcp_child.c b/src/pcp_con/pcp_child.c
index 1d61fb4b7..8f9d6a8d4 100644
--- a/src/pcp_con/pcp_child.c
+++ b/src/pcp_con/pcp_child.c
@@ -22,6 +22,9 @@
*
*/
#include "config.h"
+#include "pool.h"
+#include "utils/palloc.h"
+#include "utils/memutils.h"
#include <sys/types.h>
#include <sys/socket.h>
@@ -57,6 +60,7 @@
#include "context/pool_process_context.h"
#include "utils/pool_process_reporting.h"
#include "watchdog/wd_ext.h"
+#include "utils/elog.h"
#define MAX_FILE_LINE_LEN 512
#define MAX_USER_PASSWD_LEN 128
@@ -81,6 +85,26 @@ static volatile sig_atomic_t pcp_got_sighup = 0;
volatile sig_atomic_t pcp_wakeup_request = 0;
static volatile sig_atomic_t pcp_restart_request = 0;
+static void do_pcp_flush(PCP_CONNECTION *frontend);
+
+static void inform_process_count(PCP_CONNECTION *frontend);
+static void inform_process_info(PCP_CONNECTION *frontend, char *buf);
+static void inform_systemDB_info(PCP_CONNECTION *frontend);
+static void inform_watchdog_info(PCP_CONNECTION *frontend, char *buf);
+static void inform_node_info(PCP_CONNECTION *frontend, char *buf);
+static void inform_node_count(PCP_CONNECTION *frontend);
+static void process_detach_node(PCP_CONNECTION *frontend,char *buf,char tos);
+static void process_attach_node(PCP_CONNECTION *frontend,char *buf);
+static void process_recovery_request(PCP_CONNECTION *frontend,char *buf);
+static void process_status_request(PCP_CONNECTION *frontend);
+static void process_promote_node(PCP_CONNECTION *frontend,char *buf, char tos);
+static void process_shutown_request(char mode);
+
+static void process_authentication(PCP_CONNECTION *frontend, char *buf, char *pcp_conf_file,
+ char* salt, int *random_salt, int *authenticated);
+static void send_md5salt(PCP_CONNECTION *frontend, char* salt);
+
+
#define CHECK_RESTART_REQUEST \
do { \
if (pcp_restart_request) \
@@ -97,12 +121,14 @@ static volatile sig_atomic_t pcp_restart_request = 0;
void
pcp_do_child(int unix_fd, int inet_fd, char *pcp_conf_file)
{
+ sigjmp_buf local_sigjmp_buf;
+ MemoryContext PCPMemoryContext;
+
PCP_CONNECTION *frontend = NULL;
int authenticated = 0;
struct timeval uptime;
char salt[4];
int random_salt = 0;
- int i;
char tos;
int rsize;
char *buf = NULL;
@@ -128,6 +154,14 @@ pcp_do_child(int unix_fd, int inet_fd, char *pcp_conf_file)
signal(SIGUSR2, wakeup_handler);
signal(SIGPIPE, SIG_IGN);
signal(SIGALRM, SIG_IGN);
+ /* Create per loop iteration memory context */
+ PCPMemoryContext = AllocSetContextCreate(TopMemoryContext,
+ "PCP_main_loop",
+ ALLOCSET_DEFAULT_MINSIZE,
+ ALLOCSET_DEFAULT_INITSIZE,
+ ALLOCSET_DEFAULT_MAXSIZE);
+
+ MemoryContextSwitchTo(TopMemoryContext);
/* Initialize my backend status */
pool_initialize_private_backend_status();
@@ -135,8 +169,26 @@ pcp_do_child(int unix_fd, int inet_fd, char *pcp_conf_file)
/* Initialize process context */
pool_init_process_context();
+ if (sigsetjmp(local_sigjmp_buf, 1) != 0)
+ {
+ pool_signal(SIGALRM, SIG_IGN);
+ if(frontend)
+ pcp_close(frontend);
+ frontend = NULL;
+ authenticated = 0;
+ error_context_stack = NULL;
+ EmitErrorReport();
+ MemoryContextSwitchTo(TopMemoryContext);
+ FlushErrorState();
+ }
+ /* We can now handle ereport(ERROR) */
+ PG_exception_stack = &local_sigjmp_buf;
+
for(;;)
{
+ MemoryContextSwitchTo(PCPMemoryContext);
+ MemoryContextResetAndDeleteChildren(PCPMemoryContext);
+
errno = 0;
CHECK_RESTART_REQUEST;
@@ -155,63 +207,51 @@ pcp_do_child(int unix_fd, int inet_fd, char *pcp_conf_file)
{
if (errorcode == TIMEOUTERR)
{
- pool_debug("pcp_child: pcp_read() has timed out");
- authenticated = 0;
- pcp_close(frontend); frontend = NULL;
- free(buf); buf = NULL;
- continue;
+ ereport(ERROR,
+ (errmsg("unable to read from client"),
+ errdetail("pcp_read time out")));
}
- pool_error("pcp_child: pcp_read() failed. reason: %s", strerror(errno));
- exit(1);
+ ereport(FATAL,
+ (errmsg("unable to read from client"),
+ errdetail("pcp_read failed with error : \"%s\"",strerror(errno))));
+
}
if (pcp_read(frontend, &rsize, sizeof(int)))
{
if (errorcode == TIMEOUTERR)
{
- pool_debug("pcp_child: pcp_read() has timed out");
- authenticated = 0;
- pcp_close(frontend); frontend = NULL;
- free(buf); buf = NULL;
- continue;
+ ereport(ERROR,
+ (errmsg("unable to read from client"),
+ errdetail("pcp_read time out")));
}
- pool_error("pcp_child: pcp_read() failed. reason: %s", strerror(errno));
- exit(1);
+ ereport(FATAL,
+ (errmsg("unable to read from client"),
+ errdetail("pcp_read failed with error : \"%s\"",strerror(errno))));
}
rsize = ntohl(rsize);
if ((rsize - sizeof(int)) > 0)
{
- buf = (char *)malloc(rsize - sizeof(int));
- if (buf == NULL)
- {
- pool_error("pcp_child: malloc() failed. reason: %s", strerror(errno));
- exit(1);
- }
+ buf = (char *)palloc(rsize - sizeof(int));
if (pcp_read(frontend, buf, rsize - sizeof(int)))
{
if (errorcode == TIMEOUTERR)
{
- pool_debug("pcp_child: pcp_read() has timed out");
- authenticated = 0;
- pcp_close(frontend); frontend = NULL;
- free(buf); buf = NULL;
- continue;
+ ereport(ERROR,
+ (errmsg("unable to read from client"),
+ errdetail("pcp_read time out")));
}
- pool_error("pcp_child: pcp_read() failed. reason: %s", strerror(errno));
- exit(1);
+ ereport(FATAL,
+ (errmsg("unable to read from client"),
+ errdetail("pcp_read failed with error : \"%s\"",strerror(errno))));
}
}
/* is this connection authenticated? if not disconnect immediately*/
if ((! authenticated) && (tos != 'R' && tos != 'M'))
{
- pool_debug("pcp_child: connection not authorized");
- free(buf);
- buf = NULL;
- pcp_close(frontend);
- frontend = NULL;
- pool_signal(SIGALRM, SIG_IGN);
- continue;
+ ereport(ERROR,
+ (errmsg("connection not authorized")));
}
/* process a request */
@@ -248,874 +288,56 @@ pcp_do_child(int unix_fd, int inet_fd, char *pcp_conf_file)
switch (tos)
{
case 'R': /* authentication */
- {
- int wsize;
-
- if (random_salt)
- {
- authenticated = user_authenticate(buf, pcp_conf_file, salt, 4);
- }
- if (!random_salt || !authenticated)
- {
- char code[] = "AuthenticationFailed";
- char mesg[] = "username and/or password do not match";
- pcp_write(frontend, "r", 1);
- wsize = htonl(sizeof(code) + sizeof(mesg) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- pcp_write(frontend, mesg, sizeof(mesg));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- pool_error("pcp_child: authentication failed");
- pcp_close(frontend);
- frontend = NULL;
- random_salt = 0;
- }
- else
- {
- char code[] = "AuthenticationOK";
- pcp_write(frontend, "r", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
- random_salt = 0;
-
- pool_debug("pcp_child: authentication OK");
- }
+ process_authentication(frontend, buf, pcp_conf_file,
+ salt, &random_salt, &authenticated);
break;
- }
case 'M': /* md5 salt */
- {
- int wsize;
-
- pool_random_salt(salt);
+ send_md5salt(frontend, salt);
random_salt = 1;
-
- pcp_write(frontend, "m", 1);
- wsize = htonl(sizeof(int) + 4);
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, salt, 4);
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- pool_debug("pcp_child: salt sent to the client");
break;
- }
case 'L': /* node count */
- {
- int wsize;
- int node_count = pool_get_node_count();
-
- char code[] = "CommandComplete";
- char mesg[16];
-
- snprintf(mesg, sizeof(mesg), "%d", node_count);
-
- pcp_write(frontend, "l", 1);
- wsize = htonl(sizeof(code) +
- strlen(mesg)+1 +
- sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- pcp_write(frontend, mesg, strlen(mesg)+1);
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- pool_debug("pcp_child: %d node(s) found", node_count);
+ inform_node_count(frontend);
break;
- }
case 'I': /* node info */
- {
- int node_id;
- int wsize;
-
- BackendInfo *bi = NULL;
-
- node_id = atoi(buf);
- bi = pool_get_node_info(node_id);
-
- if (bi == NULL) {
- char code[] = "Invalid Node ID";
-
- pcp_write(frontend, "e", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- pool_debug("pcp_child: invalid node ID");
- }
- else
- {
- char code[] = "CommandComplete";
- char port_str[6];
- char status[2];
- char weight_str[20];
-
- snprintf(port_str, sizeof(port_str), "%d", bi->backend_port);
- snprintf(status, sizeof(status), "%d", bi->backend_status);
- snprintf(weight_str, sizeof(weight_str), "%f", bi->backend_weight);
-
- pcp_write(frontend, "i", 1);
- wsize = htonl(sizeof(code) +
- strlen(bi->backend_hostname)+1 +
- strlen(port_str)+1 +
- strlen(status)+1 +
- strlen(weight_str)+1 +
- sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- pcp_write(frontend, bi->backend_hostname, strlen(bi->backend_hostname)+1);
- pcp_write(frontend, port_str, strlen(port_str)+1);
- pcp_write(frontend, status, strlen(status)+1);
- pcp_write(frontend, weight_str, strlen(weight_str)+1);
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- pool_debug("pcp_child: retrieved node information from shared memory");
- }
+ inform_node_info(frontend, buf);
break;
- }
-
case 'N': /* process count */
- {
- int wsize;
- int process_count;
- char process_count_str[16];
- int *process_list = NULL;
- char code[] = "CommandComplete";
- char *mesg = NULL;
- int i;
- int total_port_len = 0;
-
- process_list = pool_get_process_list(&process_count);
-
- mesg = (char *)malloc(6*process_count); /* port# is at most 5 characters long (MAX:65535) */
- if (mesg == NULL)
- {
- pool_error("pcp_child: malloc() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- snprintf(process_count_str, sizeof(process_count_str), "%d", process_count);
-
- for (i = 0; i < process_count; i++)
- {
- char port[6];
- snprintf(port, sizeof(port), "%d", process_list[i]);
- snprintf(mesg+total_port_len, strlen(port)+1, "%s", port);
- total_port_len += strlen(port)+1;
- }
-
- pcp_write(frontend, "n", 1);
- wsize = htonl(sizeof(code) +
- strlen(process_count_str)+1 +
- total_port_len +
- sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- pcp_write(frontend, process_count_str, strlen(process_count_str)+1);
- pcp_write(frontend, mesg, total_port_len);
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- free(process_list);
- free(mesg);
-
- pool_debug("pcp_child: %d process(es) found", process_count);
+ inform_process_count(frontend);
break;
- }
-
case 'P': /* process info */
- {
- int proc_id;
- int wsize;
- int num_proc = pool_config->num_init_children;
- int i;
-
- proc_id = atoi(buf);
-
- if ((proc_id != 0) && (pool_get_process_info(proc_id) == NULL))
- {
- char code[] = "InvalidProcessID";
-
- pcp_write(frontend, "e", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- pool_debug("pcp_child: invalid process ID");
- }
- else
- {
- /* First, send array size of connection_info */
- char arr_code[] = "ArraySize";
- char con_info_size[16];
- /* Finally, indicate that all data is sent */
- char fin_code[] = "CommandComplete";
-
- POOL_REPORT_POOLS *pools = get_pools(&num_proc);
-
- if (proc_id == 0)
- {
- snprintf(con_info_size, sizeof(con_info_size), "%d", num_proc);
- }
- else
- {
- snprintf(con_info_size, sizeof(con_info_size), "%d", pool_config->max_pool * NUM_BACKENDS);
- }
-
- pcp_write(frontend, "p", 1);
- wsize = htonl(sizeof(arr_code) +
- strlen(con_info_size)+1 +
- sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, arr_code, sizeof(arr_code));
- pcp_write(frontend, con_info_size, strlen(con_info_size)+1);
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- free(pools);
- exit(1);
- }
-
- /* Second, send process information for all connection_info */
- for (i=0; i<num_proc; i++)
- {
- char code[] = "ProcessInfo";
- char proc_pid[16];
- char proc_start_time[20];
- char proc_create_time[20];
- char majorversion[5];
- char minorversion[5];
- char pool_counter[16];
- char backend_id[16];
- char backend_pid[16];
- char connected[2];
-
- if (proc_id != 0 && proc_id != pools[i].pool_pid) continue;
-
- snprintf(proc_pid, sizeof(proc_pid), "%d", pools[i].pool_pid);
- snprintf(proc_start_time, sizeof(proc_start_time), "%ld", pools[i].start_time);
- snprintf(proc_create_time, sizeof(proc_create_time), "%ld", pools[i].create_time);
- snprintf(majorversion, sizeof(majorversion), "%d", pools[i].pool_majorversion);
- snprintf(minorversion, sizeof(minorversion), "%d", pools[i].pool_minorversion);
- snprintf(pool_counter, sizeof(pool_counter), "%d", pools[i].pool_counter);
- snprintf(backend_id, sizeof(backend_pid), "%d", pools[i].backend_id);
- snprintf(backend_pid, sizeof(backend_pid), "%d", pools[i].pool_backendpid);
- snprintf(connected, sizeof(connected), "%d", pools[i].pool_connected);
-
- pcp_write(frontend, "p", 1);
- wsize = htonl( sizeof(code) +
- strlen(proc_pid)+1 +
- strlen(pools[i].database)+1 +
- strlen(pools[i].username)+1 +
- strlen(proc_start_time)+1 +
- strlen(proc_create_time)+1 +
- strlen(majorversion)+1 +
- strlen(minorversion)+1 +
- strlen(pool_counter)+1 +
- strlen(backend_id)+1 +
- strlen(backend_pid)+1 +
- strlen(connected)+1 +
- sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- pcp_write(frontend, proc_pid, strlen(proc_pid)+1);
- pcp_write(frontend, pools[i].database, strlen(pools[i].database)+1);
- pcp_write(frontend, pools[i].username, strlen(pools[i].username)+1);
- pcp_write(frontend, proc_start_time, strlen(proc_start_time)+1);
- pcp_write(frontend, proc_create_time, strlen(proc_create_time)+1);
- pcp_write(frontend, majorversion, strlen(majorversion)+1);
- pcp_write(frontend, minorversion, strlen(minorversion)+1);
- pcp_write(frontend, pool_counter, strlen(pool_counter)+1);
- pcp_write(frontend, backend_id, strlen(backend_id)+1);
- pcp_write(frontend, backend_pid, strlen(backend_pid)+1);
- pcp_write(frontend, connected, strlen(connected)+1);
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- free(pools);
- exit(1);
- }
- }
-
- pcp_write(frontend, "p", 1);
- wsize = htonl(sizeof(fin_code) +
- sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, fin_code, sizeof(fin_code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- free(pools);
- exit(1);
- }
-
- pool_debug("pcp_child: retrieved process information from shared memory");
- free(pools);
- }
+ inform_process_info(frontend, buf);
break;
- }
-
case 'S': /* SystemDB info */
- {
- int wsize;
-
- SystemDBInfo *si = NULL;
- si = pool_get_system_db_info();
-
- if (si == NULL)
- {
- char code[] = "SystemDBNotDefined";
-
- pcp_write(frontend, "e", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
- }
- else
- {
- /* first, send systemDB information */
- char code[] = "SystemDBInfo";
- char port[6];
- char status[2];
- char dist_def_num[16];
- /* finally, indicate that all data is sent */
- char fin_code[] = "CommandComplete";
-
- /* since PCP clients can only see SystemDBInfo, set system_db_status from the shared memory */
- si->system_db_status = SYSDB_STATUS;
-
- snprintf(port, sizeof(port), "%d", si->port);
- snprintf(status, sizeof(status), "%d", si->system_db_status);
- snprintf(dist_def_num, sizeof(dist_def_num), "%d", si->dist_def_num);
-
- pcp_write(frontend, "s", 1);
- wsize = htonl(sizeof(code) +
- strlen(si->hostname)+1 +
- strlen(port)+1 +
- strlen(si->user)+1 +
- strlen(si->password)+1 +
- strlen(si->schema_name)+1 +
- strlen(si->database_name)+1 +
- strlen(dist_def_num)+1 +
- strlen(status)+1 +
- sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- pcp_write(frontend, si->hostname, strlen(si->hostname)+1);
- pcp_write(frontend, port, strlen(port)+1);
- pcp_write(frontend, si->user, strlen(si->user)+1);
- pcp_write(frontend, si->password, strlen(si->password)+1);
- pcp_write(frontend, si->schema_name, strlen(si->schema_name)+1);
- pcp_write(frontend, si->database_name, strlen(si->database_name)+1);
- pcp_write(frontend, dist_def_num, strlen(dist_def_num)+1);
- pcp_write(frontend, status, strlen(status)+1);
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- /* second, send DistDefInfo if any */
- if (si->dist_def_num > 0)
- {
- char dist_code[] = "DistDefInfo";
- char col_num[16];
- int col_list_total_len;
- int type_list_total_len;
- char *col_list = NULL;
- char *type_list = NULL;
- int col_list_offset;
- int type_list_offset;
- DistDefInfo *ddi;
- int i, j;
-
- for (i = 0; i < si->dist_def_num; i++)
- {
- ddi = &si->dist_def_slot[i];
- snprintf(col_num, sizeof(col_num), "%d", ddi->col_num);
-
- col_list_total_len = type_list_total_len = 0;
- for (j = 0; j < ddi->col_num; j++)
- {
- col_list_total_len += strlen(ddi->col_list[j]) + 1;
- type_list_total_len += strlen(ddi->type_list[j]) + 1;
- }
-
- col_list = (char *)malloc(col_list_total_len);
- type_list = (char *)malloc(type_list_total_len);
- if (col_list == NULL || type_list == NULL)
- {
- pool_error("pcp_child: malloc() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- col_list_offset = type_list_offset = 0;
- for (j = 0; j < ddi->col_num; j++)
- {
- snprintf(col_list + col_list_offset, strlen(ddi->col_list[j])+1,
- "%s", ddi->col_list[j]);
- snprintf(type_list + type_list_offset, strlen(ddi->type_list[j])+1,
- "%s", ddi->type_list[j]);
- col_list_offset += strlen(ddi->col_list[j]) + 1;
- type_list_offset += strlen(ddi->type_list[j]) + 1;
- }
-
- pcp_write(frontend, "s", 1);
- wsize = htonl(sizeof(dist_code) +
- strlen(ddi->dbname)+1 +
- strlen(ddi->schema_name)+1 +
- strlen(ddi->table_name)+1 +
- strlen(ddi->dist_key_col_name)+1 +
- strlen(col_num)+1 +
- col_list_total_len +
- type_list_total_len +
- strlen(ddi->dist_def_func)+1 +
- sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, dist_code, sizeof(dist_code));
- pcp_write(frontend, ddi->dbname, strlen(ddi->dbname)+1);
- pcp_write(frontend, ddi->schema_name, strlen(ddi->schema_name)+1);
- pcp_write(frontend, ddi->table_name, strlen(ddi->table_name)+1);
- pcp_write(frontend, ddi->dist_key_col_name, strlen(ddi->dist_key_col_name)+1);
- pcp_write(frontend, col_num, strlen(col_num)+1);
- pcp_write(frontend, col_list, col_list_total_len);
- pcp_write(frontend, type_list, type_list_total_len);
- pcp_write(frontend, ddi->dist_def_func, strlen(ddi->dist_def_func)+1);
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- free(col_list);
- free(type_list);
- }
- }
-
- pcp_write(frontend, "s", 1);
- wsize = htonl(sizeof(fin_code) +
- sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, fin_code, sizeof(fin_code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- pool_debug("pcp_child: retrieved SystemDB information from shared memory");
- }
+ inform_systemDB_info(frontend);
break;
- }
-
case 'W': /* watchdog info */
- {
- int wd_index;
- int wsize;
-
- WdInfo *wi = NULL;
-
- if (!pool_config->use_watchdog)
- {
- char code[] = "watchdog not enabled";
-
- pcp_write(frontend, "e", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- pool_debug("pcp_child: watcdhog not enabled");
- break;
- }
-
- wd_index = atoi(buf);
- wi = wd_get_watchdog_info(wd_index);
-
- if (wi == NULL) {
- char code[] = "Invalid watchdog index";
-
- pcp_write(frontend, "e", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- pool_debug("pcp_child: invalid watchdog index");
- }
- else
- {
- char code[] = "CommandComplete";
- char pgpool_port_str[6];
- char wd_port_str[6];
- char status[2];
-
- snprintf(pgpool_port_str, sizeof(pgpool_port_str), "%d", wi->pgpool_port);
- snprintf(wd_port_str, sizeof(wd_port_str), "%d", wi->wd_port);
- snprintf(status, sizeof(status), "%d", wi->status);
-
- pcp_write(frontend, "w", 1);
- wsize = htonl(sizeof(code) +
- strlen(wi->hostname)+1 +
- strlen(pgpool_port_str)+1 +
- strlen(wd_port_str)+1 +
- strlen(status)+1 +
- sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
-
- pcp_write(frontend, wi->hostname, strlen(wi->hostname)+1);
- pcp_write(frontend, pgpool_port_str, strlen(pgpool_port_str)+1);
- pcp_write(frontend, wd_port_str, strlen(wd_port_str)+1);
- pcp_write(frontend, status, strlen(status)+1);
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- pool_debug("pcp_child: retrieved node information from shared memory");
- }
+ inform_watchdog_info(frontend, buf);
break;
- }
-
case 'D': /* detach node */
case 'd': /* detach node gracefully */
- {
- int node_id;
- int wsize;
- char code[] = "CommandComplete";
- bool gracefully;
-
- if (tos == 'D')
- gracefully = false;
- else
- gracefully = true;
-
- node_id = atoi(buf);
- pool_debug("pcp_child: detaching Node ID %d", node_id);
- pool_detach_node(node_id, gracefully);
-
- pcp_write(frontend, "d", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
+ process_detach_node(frontend, buf, tos);
break;
- }
-
case 'C': /* attach node */
- {
- int node_id;
- int wsize;
- char code[] = "CommandComplete";
-
- node_id = atoi(buf);
- pool_debug("pcp_child: attaching Node ID %d", node_id);
- send_failback_request(node_id);
-
- pcp_write(frontend, "c", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
+ process_attach_node(frontend, buf);
break;
- }
-
case 'T':
- {
- char mode = buf[0];
- pid_t ppid = getppid();
-
- if (mode == 's')
- {
- pool_debug("pcp_child: sending SIGTERM to the parent process(%d)", ppid);
- kill(ppid, SIGTERM);
- }
- else if (mode == 'f')
- {
- pool_debug("pcp_child: sending SIGINT to the parent process(%d)", ppid);
- kill(ppid, SIGINT);
- }
- else if (mode == 'i')
- {
- pool_debug("pcp_child: sending SIGQUIT to the parent process(%d)", ppid);
- kill(ppid, SIGQUIT);
- }
- else
- {
- pool_debug("pcp_child: invalid shutdown mode %c", mode);
- }
-
+ process_shutown_request(buf[0]);
break;
- }
case 'O': /* recovery request */
- {
- int node_id;
- int wsize;
- char code[] = "CommandComplete";
- int r;
-
- node_id = atoi(buf);
-
- if ( (node_id < 0) || (node_id >= pool_config->backend_desc->num_backends) )
- {
- char code[] = "NodeIdOutOfRange";
- pool_error("pcp_child: node id %d is not valid", node_id);
- pcp_write(frontend, "e", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
- exit(1);
- }
-
- if ((!REPLICATION &&
- !(MASTER_SLAVE &&
- !strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP))) ||
- (MASTER_SLAVE &&
- !strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP) &&
- node_id == PRIMARY_NODE_ID))
- {
- int len;
- char *msg;
-
- if (MASTER_SLAVE && !strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP))
- msg = "primary server cannot be recovered by online recovery.";
- else
- msg = "recovery request is accepted only in replication mode or stereaming replication mode. ";
-
- len = strlen(msg)+1;
- pcp_write(frontend, "e", 1);
- wsize = htonl(sizeof(int) + len);
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, msg, len);
- }
- else
- {
- pool_debug("pcp_child: start online recovery");
-
- r = start_recovery(node_id);
- finish_recovery();
-
- if (r == 0) /* success */
- {
- pcp_write(frontend, "c", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- }
- else
- {
- int len = strlen("recovery failed") + 1;
- pcp_write(frontend, "e", 1);
- wsize = htonl(sizeof(int) + len);
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, "recovery failed", len);
- }
- }
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
- }
+ process_recovery_request(frontend, buf);
break;
case 'B': /* status request*/
- {
- int nrows = 0;
- POOL_REPORT_CONFIG *status = get_config(&nrows);
- int len = 0;
- /* First, send array size of connection_info */
- char arr_code[] = "ArraySize";
- char code[] = "ProcessConfig";
- /* Finally, indicate that all data is sent */
- char fin_code[] = "CommandComplete";
-
- pcp_write(frontend, "b", 1);
- len = htonl(sizeof(arr_code) + sizeof(int) + sizeof(int));
- pcp_write(frontend, &len, sizeof(int));
- pcp_write(frontend, arr_code, sizeof(arr_code));
- len = htonl(nrows);
- pcp_write(frontend, &len, sizeof(int));
-
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- for (i = 0; i < nrows; i++)
- {
- pcp_write(frontend, "b", 1);
- len = htonl(sizeof(int)
- + sizeof(code)
- + strlen(status[i].name) + 1
- + strlen(status[i].value) + 1
- + strlen(status[i].desc) + 1
- );
-
- pcp_write(frontend, &len, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- pcp_write(frontend, status[i].name, strlen(status[i].name)+1);
- pcp_write(frontend, status[i].value, strlen(status[i].value)+1);
- pcp_write(frontend, status[i].desc, strlen(status[i].desc)+1);
- }
-
- pcp_write(frontend, "b", 1);
- len = htonl(sizeof(fin_code) + sizeof(int));
- pcp_write(frontend, &len, sizeof(int));
- pcp_write(frontend, fin_code, sizeof(fin_code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
-
- free(status);
-
- pool_debug("pcp_child: retrieved status information");
+ process_status_request(frontend);
break;
- }
case 'J': /* promote node */
case 'j': /* promote node gracefully */
- {
- int node_id;
- int wsize;
- char code[] = "CommandComplete";
- bool gracefully;
-
- if (tos == 'J')
- gracefully = false;
- else
- gracefully = true;
-
- node_id = atoi(buf);
- if ( (node_id < 0) || (node_id >= pool_config->backend_desc->num_backends) )
- {
- char code[] = "NodeIdOutOfRange";
- pool_error("pcp_child: node id %d is not valid", node_id);
- pcp_write(frontend, "e", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
- exit(1);
- }
- /* promoting node is reserved to Streaming Replication */
- if (!MASTER_SLAVE || (strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP) != 0))
- {
- char code[] = "NotInStreamingReplication";
- pool_error("pcp_child: not in streaming replication mode, can't promote node id %d", node_id);
- pcp_write(frontend, "e", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
- exit(1);
- }
-
- if (node_id == PRIMARY_NODE_ID)
- {
- char code[] = "NodeIdAlreadyPrimary";
- pool_error("pcp_child: specified node is already primary node, can't promote node id %d", node_id);
- pcp_write(frontend, "e", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
- exit(1);
- }
-
- pool_debug("pcp_child: promoting Node ID %d", node_id);
- pool_promote_node(node_id, gracefully);
-
- pcp_write(frontend, "d", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- if (pcp_flush(frontend) < 0)
- {
- pool_error("pcp_child: pcp_flush() failed. reason: %s", strerror(errno));
- exit(1);
- }
+ process_promote_node(frontend,buf,tos);
break;
- }
case 'F':
pool_debug("pcp_child: stop online recovery");
@@ -1129,13 +351,10 @@ pcp_do_child(int unix_fd, int inet_fd, char *pcp_conf_file)
break;
default:
- pool_error("pcp_child: unknown packet type %c received", tos);
- exit(1);
+ ereport(FATAL,
+ (errmsg("invalid pcp packet"),
+ errdetail("unknown pcp packet type \"%c\"",tos)));
}
-
- free(buf);
- buf = NULL;
-
/* seems ok. cancel idle check timer */
pool_signal(SIGALRM, SIG_IGN);
}
@@ -1205,9 +424,9 @@ pcp_do_accept(int unix_fd, int inet_fd)
{
if (errno == EAGAIN || errno == EINTR)
return NULL;
-
- pool_error("pcp_child: select() failed. reason: %s", strerror(errno));
- return NULL;
+ ereport(ERROR,
+ (errmsg("unable to accept new pcp connection"),
+ errdetail("select system call failed with error : \"%s\"",strerror(errno))));
}
if (FD_ISSET(unix_fd, &readmask))
@@ -1231,8 +450,9 @@ pcp_do_accept(int unix_fd, int inet_fd)
* can be silently ignored.
*/
if (errno != EAGAIN && errno != EWOULDBLOCK)
- pool_error("pcp_child: accept() failed. reason: %s", strerror(errno));
- return NULL;
+ ereport(ERROR,
+ (errmsg("unable to accept new pcp connection"),
+ errdetail("socket accept system call failed with error : \"%s\"",strerror(errno))));
}
if (pcp_got_sighup)
@@ -1251,25 +471,24 @@ pcp_do_accept(int unix_fd, int inet_fd)
(char *) &on,
sizeof(on)) < 0)
{
- pool_error("pcp_child: setsockopt() failed: %s", strerror(errno));
close(afd);
- return NULL;
+ ereport(ERROR,
+ (errmsg("unable to accept new pcp connection"),
+ errdetail("setsockopt system call failed with error : \"%s\"",strerror(errno))));
}
if (setsockopt(afd, SOL_SOCKET, SO_KEEPALIVE,
(char *) &on,
sizeof(on)) < 0)
{
- pool_error("pcp_child: setsockopt() failed: %s", strerror(errno));
close(afd);
- return NULL;
+ ereport(ERROR,
+ (errmsg("unable to accept new pcp connection"),
+ errdetail("setsockopt system call failed with error : \"%s\"",strerror(errno))));
}
}
if ((pc = pcp_open(afd)) == NULL)
- {
close(afd);
- return NULL;
- }
return pc;
}
@@ -1285,13 +504,16 @@ unset_nonblock(int fd)
var = fcntl(fd, F_GETFL, 0);
if (var == -1)
{
- pool_error("pcp_child: fcntl failed. %s", strerror(errno));
- exit(1);
+ ereport(FATAL,
+ (errmsg("unable to connect"),
+ errdetail("fcntl system call failed with error : \"%s\"",strerror(errno))));
+
}
if (fcntl(fd, F_SETFL, var & ~O_NONBLOCK) == -1)
{
- pool_error("pcp_child: fcntl failed. %s", strerror(errno));
- exit(1);
+ ereport(FATAL,
+ (errmsg("unable to connect"),
+ errdetail("fcntl system call failed with error : \"%s\"",strerror(errno))));
}
}
@@ -1491,3 +713,764 @@ static int pool_promote_node(int node_id, bool gracefully)
return 0;
}
+
+static void
+inform_process_count(PCP_CONNECTION *frontend)
+{
+ int wsize;
+ int process_count;
+ char process_count_str[16];
+ int *process_list = NULL;
+ char code[] = "CommandComplete";
+ char *mesg = NULL;
+ int i;
+ int total_port_len = 0;
+
+ process_list = pool_get_process_list(&process_count);
+
+ mesg = (char *)palloc(6*process_count); /* port# is at most 5 characters long (MAX:65535) */
+
+ snprintf(process_count_str, sizeof(process_count_str), "%d", process_count);
+
+ for (i = 0; i < process_count; i++)
+ {
+ char port[6];
+ snprintf(port, sizeof(port), "%d", process_list[i]);
+ snprintf(mesg+total_port_len, strlen(port)+1, "%s", port);
+ total_port_len += strlen(port)+1;
+ }
+
+ pcp_write(frontend, "n", 1);
+ wsize = htonl(sizeof(code) +
+ strlen(process_count_str)+1 +
+ total_port_len +
+ sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ pcp_write(frontend, process_count_str, strlen(process_count_str)+1);
+ pcp_write(frontend, mesg, total_port_len);
+ do_pcp_flush(frontend);
+
+ pfree(process_list);
+ pfree(mesg);
+
+ pool_debug("pcp_child: %d process(es) found", process_count);
+}
+
+static void
+inform_process_info(PCP_CONNECTION *frontend,char *buf)
+{
+ int proc_id;
+ int wsize;
+ int num_proc = pool_config->num_init_children;
+ int i;
+
+ proc_id = atoi(buf);
+
+ if ((proc_id != 0) && (pool_get_process_info(proc_id) == NULL))
+ {
+ char code[] = "InvalidProcessID";
+
+ pcp_write(frontend, "e", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+
+ pool_debug("pcp_child: invalid process ID");
+ }
+ else
+ {
+ /* First, send array size of connection_info */
+ char arr_code[] = "ArraySize";
+ char con_info_size[16];
+ /* Finally, indicate that all data is sent */
+ char fin_code[] = "CommandComplete";
+
+ POOL_REPORT_POOLS *pools = get_pools(&num_proc);
+
+ if (proc_id == 0)
+ {
+ snprintf(con_info_size, sizeof(con_info_size), "%d", num_proc);
+ }
+ else
+ {
+ snprintf(con_info_size, sizeof(con_info_size), "%d", pool_config->max_pool * NUM_BACKENDS);
+ }
+
+ pcp_write(frontend, "p", 1);
+ wsize = htonl(sizeof(arr_code) +
+ strlen(con_info_size)+1 +
+ sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, arr_code, sizeof(arr_code));
+ pcp_write(frontend, con_info_size, strlen(con_info_size)+1);
+ do_pcp_flush(frontend);
+
+ /* Second, send process information for all connection_info */
+ for (i=0; i<num_proc; i++)
+ {
+ char code[] = "ProcessInfo";
+ char proc_pid[16];
+ char proc_start_time[20];
+ char proc_create_time[20];
+ char majorversion[5];
+ char minorversion[5];
+ char pool_counter[16];
+ char backend_id[16];
+ char backend_pid[16];
+ char connected[2];
+
+ if (proc_id != 0 && proc_id != pools[i].pool_pid) continue;
+
+ snprintf(proc_pid, sizeof(proc_pid), "%d", pools[i].pool_pid);
+ snprintf(proc_start_time, sizeof(proc_start_time), "%ld", pools[i].start_time);
+ snprintf(proc_create_time, sizeof(proc_create_time), "%ld", pools[i].create_time);
+ snprintf(majorversion, sizeof(majorversion), "%d", pools[i].pool_majorversion);
+ snprintf(minorversion, sizeof(minorversion), "%d", pools[i].pool_minorversion);
+ snprintf(pool_counter, sizeof(pool_counter), "%d", pools[i].pool_counter);
+ snprintf(backend_id, sizeof(backend_pid), "%d", pools[i].backend_id);
+ snprintf(backend_pid, sizeof(backend_pid), "%d", pools[i].pool_backendpid);
+ snprintf(connected, sizeof(connected), "%d", pools[i].pool_connected);
+
+ pcp_write(frontend, "p", 1);
+ wsize = htonl( sizeof(code) +
+ strlen(proc_pid)+1 +
+ strlen(pools[i].database)+1 +
+ strlen(pools[i].username)+1 +
+ strlen(proc_start_time)+1 +
+ strlen(proc_create_time)+1 +
+ strlen(majorversion)+1 +
+ strlen(minorversion)+1 +
+ strlen(pool_counter)+1 +
+ strlen(backend_id)+1 +
+ strlen(backend_pid)+1 +
+ strlen(connected)+1 +
+ sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ pcp_write(frontend, proc_pid, strlen(proc_pid)+1);
+ pcp_write(frontend, pools[i].database, strlen(pools[i].database)+1);
+ pcp_write(frontend, pools[i].username, strlen(pools[i].username)+1);
+ pcp_write(frontend, proc_start_time, strlen(proc_start_time)+1);
+ pcp_write(frontend, proc_create_time, strlen(proc_create_time)+1);
+ pcp_write(frontend, majorversion, strlen(majorversion)+1);
+ pcp_write(frontend, minorversion, strlen(minorversion)+1);
+ pcp_write(frontend, pool_counter, strlen(pool_counter)+1);
+ pcp_write(frontend, backend_id, strlen(backend_id)+1);
+ pcp_write(frontend, backend_pid, strlen(backend_pid)+1);
+ pcp_write(frontend, connected, strlen(connected)+1);
+ do_pcp_flush(frontend);
+ }
+
+ pcp_write(frontend, "p", 1);
+ wsize = htonl(sizeof(fin_code) +
+ sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, fin_code, sizeof(fin_code));
+ do_pcp_flush(frontend);
+ pool_debug("pcp_child: retrieved process information from shared memory");
+ pfree(pools);
+ }
+}
+
+static void
+inform_systemDB_info(PCP_CONNECTION *frontend)
+{
+ int wsize;
+
+ SystemDBInfo *si = NULL;
+ si = pool_get_system_db_info();
+
+ if (si == NULL)
+ {
+ char code[] = "SystemDBNotDefined";
+
+ pcp_write(frontend, "e", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+ }
+ else
+ {
+ /* first, send systemDB information */
+ char code[] = "SystemDBInfo";
+ char port[6];
+ char status[2];
+ char dist_def_num[16];
+ /* finally, indicate that all data is sent */
+ char fin_code[] = "CommandComplete";
+
+ /* since PCP clients can only see SystemDBInfo, set system_db_status from the shared memory */
+ si->system_db_status = SYSDB_STATUS;
+
+ snprintf(port, sizeof(port), "%d", si->port);
+ snprintf(status, sizeof(status), "%d", si->system_db_status);
+ snprintf(dist_def_num, sizeof(dist_def_num), "%d", si->dist_def_num);
+
+ pcp_write(frontend, "s", 1);
+ wsize = htonl(sizeof(code) +
+ strlen(si->hostname)+1 +
+ strlen(port)+1 +
+ strlen(si->user)+1 +
+ strlen(si->password)+1 +
+ strlen(si->schema_name)+1 +
+ strlen(si->database_name)+1 +
+ strlen(dist_def_num)+1 +
+ strlen(status)+1 +
+ sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ pcp_write(frontend, si->hostname, strlen(si->hostname)+1);
+ pcp_write(frontend, port, strlen(port)+1);
+ pcp_write(frontend, si->user, strlen(si->user)+1);
+ pcp_write(frontend, si->password, strlen(si->password)+1);
+ pcp_write(frontend, si->schema_name, strlen(si->schema_name)+1);
+ pcp_write(frontend, si->database_name, strlen(si->database_name)+1);
+ pcp_write(frontend, dist_def_num, strlen(dist_def_num)+1);
+ pcp_write(frontend, status, strlen(status)+1);
+ do_pcp_flush(frontend);
+ /* second, send DistDefInfo if any */
+ if (si->dist_def_num > 0)
+ {
+ char dist_code[] = "DistDefInfo";
+ char col_num[16];
+ int col_list_total_len;
+ int type_list_total_len;
+ char *col_list = NULL;
+ char *type_list = NULL;
+ int col_list_offset;
+ int type_list_offset;
+ DistDefInfo *ddi;
+ int i, j;
+
+ for (i = 0; i < si->dist_def_num; i++)
+ {
+ ddi = &si->dist_def_slot[i];
+ snprintf(col_num, sizeof(col_num), "%d", ddi->col_num);
+
+ col_list_total_len = type_list_total_len = 0;
+ for (j = 0; j < ddi->col_num; j++)
+ {
+ col_list_total_len += strlen(ddi->col_list[j]) + 1;
+ type_list_total_len += strlen(ddi->type_list[j]) + 1;
+ }
+
+ col_list = (char *)palloc(col_list_total_len);
+ type_list = (char *)palloc(type_list_total_len);
+
+ col_list_offset = type_list_offset = 0;
+ for (j = 0; j < ddi->col_num; j++)
+ {
+ snprintf(col_list + col_list_offset, strlen(ddi->col_list[j])+1,
+ "%s", ddi->col_list[j]);
+ snprintf(type_list + type_list_offset, strlen(ddi->type_list[j])+1,
+ "%s", ddi->type_list[j]);
+ col_list_offset += strlen(ddi->col_list[j]) + 1;
+ type_list_offset += strlen(ddi->type_list[j]) + 1;
+ }
+
+ pcp_write(frontend, "s", 1);
+ wsize = htonl(sizeof(dist_code) +
+ strlen(ddi->dbname)+1 +
+ strlen(ddi->schema_name)+1 +
+ strlen(ddi->table_name)+1 +
+ strlen(ddi->dist_key_col_name)+1 +
+ strlen(col_num)+1 +
+ col_list_total_len +
+ type_list_total_len +
+ strlen(ddi->dist_def_func)+1 +
+ sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, dist_code, sizeof(dist_code));
+ pcp_write(frontend, ddi->dbname, strlen(ddi->dbname)+1);
+ pcp_write(frontend, ddi->schema_name, strlen(ddi->schema_name)+1);
+ pcp_write(frontend, ddi->table_name, strlen(ddi->table_name)+1);
+ pcp_write(frontend, ddi->dist_key_col_name, strlen(ddi->dist_key_col_name)+1);
+ pcp_write(frontend, col_num, strlen(col_num)+1);
+ pcp_write(frontend, col_list, col_list_total_len);
+ pcp_write(frontend, type_list, type_list_total_len);
+ pcp_write(frontend, ddi->dist_def_func, strlen(ddi->dist_def_func)+1);
+ do_pcp_flush(frontend);
+
+ pfree(col_list);
+ pfree(type_list);
+ }
+ }
+
+ pcp_write(frontend, "s", 1);
+ wsize = htonl(sizeof(fin_code) +
+ sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, fin_code, sizeof(fin_code));
+ do_pcp_flush(frontend);
+
+ pool_debug("pcp_child: retrieved SystemDB information from shared memory");
+ }
+}
+
+static void
+inform_watchdog_info(PCP_CONNECTION *frontend,char *buf)
+{
+ int wd_index;
+ int wsize;
+
+ WdInfo *wi = NULL;
+
+ if (!pool_config->use_watchdog)
+ {
+ char code[] = "watchdog not enabled";
+
+ pcp_write(frontend, "e", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+
+ pool_debug("pcp_child: watcdhog not enabled");
+ return;
+ }
+
+ wd_index = atoi(buf);
+ wi = wd_get_watchdog_info(wd_index);
+
+ if (wi == NULL) {
+ char code[] = "Invalid watchdog index";
+
+ pcp_write(frontend, "e", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+
+ pool_debug("pcp_child: invalid watchdog index");
+ }
+ else
+ {
+ char code[] = "CommandComplete";
+ char pgpool_port_str[6];
+ char wd_port_str[6];
+ char status[2];
+
+ snprintf(pgpool_port_str, sizeof(pgpool_port_str), "%d", wi->pgpool_port);
+ snprintf(wd_port_str, sizeof(wd_port_str), "%d", wi->wd_port);
+ snprintf(status, sizeof(status), "%d", wi->status);
+
+ pcp_write(frontend, "w", 1);
+ wsize = htonl(sizeof(code) +
+ strlen(wi->hostname)+1 +
+ strlen(pgpool_port_str)+1 +
+ strlen(wd_port_str)+1 +
+ strlen(status)+1 +
+ sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+
+ pcp_write(frontend, wi->hostname, strlen(wi->hostname)+1);
+ pcp_write(frontend, pgpool_port_str, strlen(pgpool_port_str)+1);
+ pcp_write(frontend, wd_port_str, strlen(wd_port_str)+1);
+ pcp_write(frontend, status, strlen(status)+1);
+ do_pcp_flush(frontend);
+
+ pool_debug("pcp_child: retrieved node information from shared memory");
+ }
+}
+
+static void
+inform_node_info(PCP_CONNECTION *frontend,char *buf)
+{
+ int node_id;
+ int wsize;
+
+ BackendInfo *bi = NULL;
+
+ node_id = atoi(buf);
+ bi = pool_get_node_info(node_id);
+
+ if (bi == NULL) {
+ char code[] = "Invalid Node ID";
+
+ pcp_write(frontend, "e", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+
+ pool_debug("pcp_child: invalid node ID");
+ }
+ else
+ {
+ char code[] = "CommandComplete";
+ char port_str[6];
+ char status[2];
+ char weight_str[20];
+
+ snprintf(port_str, sizeof(port_str), "%d", bi->backend_port);
+ snprintf(status, sizeof(status), "%d", bi->backend_status);
+ snprintf(weight_str, sizeof(weight_str), "%f", bi->backend_weight);
+
+ pcp_write(frontend, "i", 1);
+ wsize = htonl(sizeof(code) +
+ strlen(bi->backend_hostname)+1 +
+ strlen(port_str)+1 +
+ strlen(status)+1 +
+ strlen(weight_str)+1 +
+ sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ pcp_write(frontend, bi->backend_hostname, strlen(bi->backend_hostname)+1);
+ pcp_write(frontend, port_str, strlen(port_str)+1);
+ pcp_write(frontend, status, strlen(status)+1);
+ pcp_write(frontend, weight_str, strlen(weight_str)+1);
+ do_pcp_flush(frontend);
+
+ pool_debug("pcp_child: retrieved node information from shared memory");
+ }
+}
+
+static void
+inform_node_count(PCP_CONNECTION *frontend)
+{
+ int wsize;
+ int node_count = pool_get_node_count();
+
+ char code[] = "CommandComplete";
+ char mesg[16];
+
+ snprintf(mesg, sizeof(mesg), "%d", node_count);
+
+ pcp_write(frontend, "l", 1);
+ wsize = htonl(sizeof(code) +
+ strlen(mesg)+1 +
+ sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ pcp_write(frontend, mesg, strlen(mesg)+1);
+ do_pcp_flush(frontend);
+
+ pool_debug("pcp_child: %d node(s) found", node_count);
+}
+
+static void
+process_detach_node(PCP_CONNECTION *frontend,char *buf, char tos)
+{
+ int node_id;
+ int wsize;
+ char code[] = "CommandComplete";
+ bool gracefully;
+
+ if (tos == 'D')
+ gracefully = false;
+ else
+ gracefully = true;
+
+ node_id = atoi(buf);
+ pool_debug("pcp_child: detaching Node ID %d", node_id);
+ pool_detach_node(node_id, gracefully);
+
+ pcp_write(frontend, "d", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+}
+
+static void
+process_attach_node(PCP_CONNECTION *frontend,char *buf)
+{
+ int node_id;
+ int wsize;
+ char code[] = "CommandComplete";
+
+ node_id = atoi(buf);
+ pool_debug("pcp_child: attaching Node ID %d", node_id);
+ send_failback_request(node_id);
+
+ pcp_write(frontend, "c", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+}
+
+
+static void
+process_recovery_request(PCP_CONNECTION *frontend,char *buf)
+{
+ int node_id;
+ int wsize;
+ char code[] = "CommandComplete";
+ int r;
+
+ node_id = atoi(buf);
+
+ if ( (node_id < 0) || (node_id >= pool_config->backend_desc->num_backends) )
+ {
+ char code[] = "NodeIdOutOfRange";
+ pcp_write(frontend, "e", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+ ereport(FATAL,
+ (errmsg("could not process recovery request"),
+ errdetail("node id %d is not valid", node_id)));
+ }
+
+ if ((!REPLICATION &&
+ !(MASTER_SLAVE &&
+ !strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP))) ||
+ (MASTER_SLAVE &&
+ !strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP) &&
+ node_id == PRIMARY_NODE_ID))
+ {
+ int len;
+ char *msg;
+
+ if (MASTER_SLAVE && !strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP))
+ msg = "primary server cannot be recovered by online recovery.";
+ else
+ msg = "recovery request is accepted only in replication mode or stereaming replication mode. ";
+
+ len = strlen(msg)+1;
+ pcp_write(frontend, "e", 1);
+ wsize = htonl(sizeof(int) + len);
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, msg, len);
+ }
+ else
+ {
+ pool_debug("pcp_child: start online recovery");
+
+ r = start_recovery(node_id);
+ finish_recovery();
+
+ if (r == 0) /* success */
+ {
+ pcp_write(frontend, "c", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ }
+ else
+ {
+ int len = strlen("recovery failed") + 1;
+ pcp_write(frontend, "e", 1);
+ wsize = htonl(sizeof(int) + len);
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, "recovery failed", len);
+ }
+ }
+ do_pcp_flush(frontend);
+}
+
+static void
+process_status_request(PCP_CONNECTION *frontend)
+{
+ int nrows = 0;
+ int i;
+ POOL_REPORT_CONFIG *status = get_config(&nrows);
+ int len = 0;
+ /* First, send array size of connection_info */
+ char arr_code[] = "ArraySize";
+ char code[] = "ProcessConfig";
+ /* Finally, indicate that all data is sent */
+ char fin_code[] = "CommandComplete";
+
+ pcp_write(frontend, "b", 1);
+ len = htonl(sizeof(arr_code) + sizeof(int) + sizeof(int));
+ pcp_write(frontend, &len, sizeof(int));
+ pcp_write(frontend, arr_code, sizeof(arr_code));
+ len = htonl(nrows);
+ pcp_write(frontend, &len, sizeof(int));
+
+ do_pcp_flush(frontend);
+
+ for (i = 0; i < nrows; i++)
+ {
+ pcp_write(frontend, "b", 1);
+ len = htonl(sizeof(int)
+ + sizeof(code)
+ + strlen(status[i].name) + 1
+ + strlen(status[i].value) + 1
+ + strlen(status[i].desc) + 1
+ );
+
+ pcp_write(frontend, &len, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ pcp_write(frontend, status[i].name, strlen(status[i].name)+1);
+ pcp_write(frontend, status[i].value, strlen(status[i].value)+1);
+ pcp_write(frontend, status[i].desc, strlen(status[i].desc)+1);
+ }
+
+ pcp_write(frontend, "b", 1);
+ len = htonl(sizeof(fin_code) + sizeof(int));
+ pcp_write(frontend, &len, sizeof(int));
+ pcp_write(frontend, fin_code, sizeof(fin_code));
+ do_pcp_flush(frontend);
+
+ pfree(status);
+
+ pool_debug("pcp_child: retrieved status information");
+}
+
+static void
+process_promote_node(PCP_CONNECTION *frontend, char *buf, char tos)
+{
+ int node_id;
+ int wsize;
+ char code[] = "CommandComplete";
+ bool gracefully;
+
+ if (tos == 'J')
+ gracefully = false;
+ else
+ gracefully = true;
+
+ node_id = atoi(buf);
+ if ( (node_id < 0) || (node_id >= pool_config->backend_desc->num_backends) )
+ {
+ char code[] = "NodeIdOutOfRange";
+ pcp_write(frontend, "e", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+ ereport(FATAL,
+ (errmsg("could not process recovery request"),
+ errdetail("node id %d is not valid", node_id)));
+ }
+ /* promoting node is reserved to Streaming Replication */
+ if (!MASTER_SLAVE || (strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP) != 0))
+ {
+ char code[] = "NotInStreamingReplication";
+ pcp_write(frontend, "e", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+ ereport(FATAL,
+ (errmsg("invalid pgpool mode for the command"),
+ errdetail("not in streaming replication mode, can't promote node id %d", node_id)));
+
+ }
+
+ if (node_id == PRIMARY_NODE_ID)
+ {
+ char code[] = "NodeIdAlreadyPrimary";
+ pcp_write(frontend, "e", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+ ereport(FATAL,
+ (errmsg("invalid pgpool mode for the command"),
+ errdetail("specified node is already primary node, can't promote node id %d", node_id)));
+
+ }
+
+ pool_debug("pcp_child: promoting Node ID %d", node_id);
+ pool_promote_node(node_id, gracefully);
+
+ pcp_write(frontend, "d", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+}
+
+static void
+process_authentication(PCP_CONNECTION *frontend, char *buf, char *pcp_conf_file,
+ char* salt, int *random_salt, int *authenticated)
+{
+ int wsize;
+
+ if (*random_salt)
+ {
+ *authenticated = user_authenticate(buf, pcp_conf_file, salt, 4);
+ }
+ if (!*random_salt || !*authenticated)
+ {
+ char code[] = "AuthenticationFailed";
+ char mesg[] = "username and/or password do not match";
+ pcp_write(frontend, "r", 1);
+ wsize = htonl(sizeof(code) + sizeof(mesg) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ pcp_write(frontend, mesg, sizeof(mesg));
+ do_pcp_flush(frontend);
+
+ *random_salt = 0;
+ ereport(ERROR,
+ (errmsg("authentication failed"),
+ errdetail("username and/or password do not match")));
+ }
+ else
+ {
+ char code[] = "AuthenticationOK";
+ pcp_write(frontend, "r", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+ *random_salt = 0;
+
+ pool_debug("pcp_child: authentication OK");
+ }
+}
+
+static void
+send_md5salt(PCP_CONNECTION *frontend, char* salt)
+{
+ int wsize;
+
+ pool_random_salt(salt);
+// random_salt = 1;
+
+ pcp_write(frontend, "m", 1);
+ wsize = htonl(sizeof(int) + 4);
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, salt, 4);
+ do_pcp_flush(frontend);
+
+ pool_debug("pcp_child: salt sent to the client");
+}
+
+static void
+process_shutown_request(char mode)
+{
+ pid_t ppid = getppid();
+
+ if (mode == 's')
+ {
+ pool_debug("pcp_child: sending SIGTERM to the parent process(%d)", ppid);
+ kill(ppid, SIGTERM);
+ }
+ else if (mode == 'f')
+ {
+ pool_debug("pcp_child: sending SIGINT to the parent process(%d)", ppid);
+ kill(ppid, SIGINT);
+ }
+ else if (mode == 'i')
+ {
+ pool_debug("pcp_child: sending SIGQUIT to the parent process(%d)", ppid);
+ kill(ppid, SIGQUIT);
+ }
+ else
+ {
+ pool_debug("pcp_child: invalid shutdown mode %c", mode);
+ }
+}
+
+/*
+ * Wrapper around pcp_flush which throws FATAL error when pcp_flush fails
+ */
+static void
+do_pcp_flush(PCP_CONNECTION *frontend)
+{
+ if (pcp_flush(frontend) < 0)
+ ereport(FATAL,
+ (errmsg("failed to flush data to client"),
+ errdetail("pcp_flush failed with error : \"%s\"",strerror(errno))));
+}
diff --git a/src/utils/pool_process_reporting.c b/src/utils/pool_process_reporting.c
index 35a1d5b09..2cbd48ae1 100644
--- a/src/utils/pool_process_reporting.c
+++ b/src/utils/pool_process_reporting.c
@@ -145,11 +145,11 @@ POOL_REPORT_CONFIG* get_config(int *nrows)
*/
#define MAXITEMS (256 + MAX_NUM_BACKENDS*4)
- POOL_REPORT_CONFIG* status = malloc(MAXITEMS * sizeof(POOL_REPORT_CONFIG));
+ POOL_REPORT_CONFIG* status = palloc0(MAXITEMS * sizeof(POOL_REPORT_CONFIG));
/* we initialize the array with NULL values so when looping
* on it, we can use it as stop condition */
- memset(status, 0, sizeof(POOL_REPORT_CONFIG) * MAXITEMS);
+// memset(status, 0, sizeof(POOL_REPORT_CONFIG) * MAXITEMS);
i = 0;
@@ -995,7 +995,7 @@ void config_reporting(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend)
send_complete_and_ready(frontend, backend, nrows);
- free(status);
+ pfree(status);
}
POOL_REPORT_NODES* get_nodes(int *nrows)
@@ -1148,7 +1148,7 @@ POOL_REPORT_POOLS* get_pools(int *nrows)
int lines = 0;
- POOL_REPORT_POOLS* pools = malloc(
+ POOL_REPORT_POOLS* pools = palloc(
pool_config->num_init_children * pool_config->max_pool * NUM_BACKENDS * sizeof(POOL_REPORT_POOLS)
);
@@ -1336,7 +1336,7 @@ void pools_reporting(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend)
send_complete_and_ready(frontend, backend, nrows);
- free(pools);
+ pfree(pools);
}
POOL_REPORT_PROCESSES* get_processes(int *nrows)
[parent: cdb49d3b783c]