pgpool2.git / summary / log / commit / refs
commit 89e6b8ba6b972886d65afa641cde95c5546d6cff
Author: Muhammad Usama <m.usama@gmail.com>
Date: Tue Feb 17 13:37:26 2015 +0000
Second part of overhauling the PCP library and utilities.
This commit contains the following changes
-- pgpool-II can now handle multiple simultaneous PCP commands.
-- pcp_child process now only listens for new connections and manages pcp_worker
children. Responsibility of authenticating PCP clients and processing of the
PCP client request is delegated to new PCP worker type process.
Which are spawned on demand when the new command request is received
-- This commit changes the way how smart shutdown request is handled by the
PCP process. Previously pcp_child was not caring about if smart or fast shutdown
request was issued, and kills the process instantly. This commit ensures that if
a smart shutdown request was received than the pcp_child stops receiving the
new PCP connections, but the current command in execution executes at full
before the processing of the shutdown request.
-- Restart pcp_child request now waits for the current command to complete
before restarting the process, instead of immediately killing the PCP process.
-- Removes all occurrences and handling of unused pcp_timeout from code and
command line arguments of PCP utilities. (Documentation changes are still a TODO)
-- Removes the global variable run_as_pcp_child which was already never used.
-- pgpool-II can now execute multiple concurrent PCP commands except for
pcp_recovery_node command.(only one pcp_recovery_node is allowed at a time).
-- Since pcp_conf_file is the global parameter so instead of passing it to the
pcp_child main process as an argument it is extended in the pcp_worker file.
-- Removes now unwanted utils/pcp/pcp_error.c and utils/pcp/pcp_timeout.c files.
-- Commit also contains some miscellaneous code cleanups.
src/Makefile.am | 3 +-
src/Makefile.in | 23 +-
src/auth/pool_hba.c | 6 +-
src/config/pool_config.c | 16 -
src/config/pool_config.l | 16 -
src/include/pcp/libpcp_ext.h | 4 -
src/include/pcp/pcp.h | 21 +-
src/include/pool.h | 21 +-
src/include/pool_config.h | 1 -
src/include/utils/pool_signal.h | 17 +-
src/libs/pcp/Makefile.am | 2 -
src/libs/pcp/Makefile.in | 7 +-
src/libs/pcp/pcp.c | 9 -
src/main/pgpool_main.c | 17 +-
src/main/pool_globals.c | 2 -
src/pcp_con/pcp_child.c | 1522 +++-----------------
src/pcp_con/pcp_worker.c | 1419 ++++++++++++++++++
src/pcp_con/recovery.c | 8 +-
src/protocol/child.c | 18 +-
src/protocol/pool_connection_pool.c | 6 +-
src/protocol/pool_process_query.c | 6 +-
src/query_cache/pool_memqcache.c | 36 +-
src/test/pgpool_setup | 4 +-
src/test/regression/tests/054.postgres_fdw/test.sh | 2 +-
src/tools/pcp/pcp_attach_node.c | 28 +-
src/tools/pcp/pcp_detach_node.c | 28 +-
src/tools/pcp/pcp_node_count.c | 27 +-
src/tools/pcp/pcp_node_info.c | 27 +-
src/tools/pcp/pcp_pool_status.c | 27 +-
src/tools/pcp/pcp_proc_count.c | 26 +-
src/tools/pcp/pcp_proc_info.c | 30 +-
src/tools/pcp/pcp_promote_node.c | 28 +-
src/tools/pcp/pcp_recovery_node.c | 28 +-
src/tools/pcp/pcp_stop_pgpool.c | 35 +-
src/tools/pcp/pcp_systemdb_info.c | 24 +-
src/tools/pcp/pcp_watchdog_info.c | 30 +-
src/tools/pgmd5/pool_config.c | 117 +-
src/utils/error/elog.c | 3 +
src/utils/pcp/pcp_error.c | 87 --
src/utils/pcp/pcp_stream.c | 36 +-
src/utils/pcp/pcp_timeout.c | 12 -
src/utils/pool_process_reporting.c | 6 +-
src/utils/pool_signal.c | 5 +-
src/watchdog/test/stab.c | 7 +-
44 files changed, 1927 insertions(+), 1870 deletions(-)
diff --git a/src/Makefile.am b/src/Makefile.am
index 04cda7a38..0076f6127 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -8,6 +8,7 @@ pgpool_SOURCES = main/main.c \
main/pgpool_main.c \
config/pool_config.l \
pcp_con/pcp_child.c \
+ pcp_con/pcp_worker.c \
pcp_con/recovery.c \
auth/md5.c \
auth/pool_auth.c \
@@ -48,8 +49,6 @@ pgpool_SOURCES = main/main.c \
utils/error/elog.c \
utils/error/assert.c \
utils/pcp/pcp_stream.c \
- utils/pcp/pcp_error.c \
- utils/pcp/pcp_timeout.c \
utils/regex_array.c \
utils/statistics.c
diff --git a/src/Makefile.in b/src/Makefile.in
index c3a50b0d9..7806a1b0f 100644
--- a/src/Makefile.in
+++ b/src/Makefile.in
@@ -104,11 +104,11 @@ PROGRAMS = $(bin_PROGRAMS)
am__dirstamp = $(am__leading_dot)dirstamp
am_pgpool_OBJECTS = main/main.$(OBJEXT) main/pool_globals.$(OBJEXT) \
main/pgpool_main.$(OBJEXT) config/pool_config.$(OBJEXT) \
- pcp_con/pcp_child.$(OBJEXT) pcp_con/recovery.$(OBJEXT) \
- auth/md5.$(OBJEXT) auth/pool_auth.$(OBJEXT) \
- auth/pool_passwd.$(OBJEXT) auth/pool_hba.$(OBJEXT) \
- protocol/pool_proto2.$(OBJEXT) protocol/child.$(OBJEXT) \
- protocol/pool_process_query.$(OBJEXT) \
+ pcp_con/pcp_child.$(OBJEXT) pcp_con/pcp_worker.$(OBJEXT) \
+ pcp_con/recovery.$(OBJEXT) auth/md5.$(OBJEXT) \
+ auth/pool_auth.$(OBJEXT) auth/pool_passwd.$(OBJEXT) \
+ auth/pool_hba.$(OBJEXT) protocol/pool_proto2.$(OBJEXT) \
+ protocol/child.$(OBJEXT) protocol/pool_process_query.$(OBJEXT) \
protocol/pool_connection_pool.$(OBJEXT) \
protocol/pool_proto_modules.$(OBJEXT) \
query_cache/pool_memqcache.$(OBJEXT) \
@@ -131,7 +131,6 @@ am_pgpool_OBJECTS = main/main.$(OBJEXT) main/pool_globals.$(OBJEXT) \
utils/getopt_long.$(OBJEXT) utils/mmgr/mcxt.$(OBJEXT) \
utils/mmgr/aset.$(OBJEXT) utils/error/elog.$(OBJEXT) \
utils/error/assert.$(OBJEXT) utils/pcp/pcp_stream.$(OBJEXT) \
- utils/pcp/pcp_error.$(OBJEXT) utils/pcp/pcp_timeout.$(OBJEXT) \
utils/regex_array.$(OBJEXT) utils/statistics.$(OBJEXT)
pgpool_OBJECTS = $(am_pgpool_OBJECTS)
pgpool_DEPENDENCIES = parser/libsql-parser.a parser/nodes.o \
@@ -420,6 +419,7 @@ pgpool_SOURCES = main/main.c \
main/pgpool_main.c \
config/pool_config.l \
pcp_con/pcp_child.c \
+ pcp_con/pcp_worker.c \
pcp_con/recovery.c \
auth/md5.c \
auth/pool_auth.c \
@@ -460,8 +460,6 @@ pgpool_SOURCES = main/main.c \
utils/error/elog.c \
utils/error/assert.c \
utils/pcp/pcp_stream.c \
- utils/pcp/pcp_error.c \
- utils/pcp/pcp_timeout.c \
utils/regex_array.c \
utils/statistics.c
@@ -671,6 +669,8 @@ pcp_con/$(DEPDIR)/$(am__dirstamp):
@: > pcp_con/$(DEPDIR)/$(am__dirstamp)
pcp_con/pcp_child.$(OBJEXT): pcp_con/$(am__dirstamp) \
pcp_con/$(DEPDIR)/$(am__dirstamp)
+pcp_con/pcp_worker.$(OBJEXT): pcp_con/$(am__dirstamp) \
+ pcp_con/$(DEPDIR)/$(am__dirstamp)
pcp_con/recovery.$(OBJEXT): pcp_con/$(am__dirstamp) \
pcp_con/$(DEPDIR)/$(am__dirstamp)
auth/$(am__dirstamp):
@@ -826,10 +826,6 @@ utils/pcp/$(DEPDIR)/$(am__dirstamp):
@: > utils/pcp/$(DEPDIR)/$(am__dirstamp)
utils/pcp/pcp_stream.$(OBJEXT): utils/pcp/$(am__dirstamp) \
utils/pcp/$(DEPDIR)/$(am__dirstamp)
-utils/pcp/pcp_error.$(OBJEXT): utils/pcp/$(am__dirstamp) \
- utils/pcp/$(DEPDIR)/$(am__dirstamp)
-utils/pcp/pcp_timeout.$(OBJEXT): utils/pcp/$(am__dirstamp) \
- utils/pcp/$(DEPDIR)/$(am__dirstamp)
utils/regex_array.$(OBJEXT): utils/$(am__dirstamp) \
utils/$(DEPDIR)/$(am__dirstamp)
utils/statistics.$(OBJEXT): utils/$(am__dirstamp) \
@@ -874,6 +870,7 @@ distclean-compile:
@AMDEP_TRUE@@am__include@ @am__quote@parallel_query/$(DEPDIR)/pool_rewrite_outfuncs.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@parallel_query/$(DEPDIR)/pool_rewrite_query.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@pcp_con/$(DEPDIR)/pcp_child.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@pcp_con/$(DEPDIR)/pcp_worker.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@pcp_con/$(DEPDIR)/recovery.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@protocol/$(DEPDIR)/child.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@protocol/$(DEPDIR)/pool_connection_pool.Po@am__quote@
@@ -906,9 +903,7 @@ distclean-compile:
@AMDEP_TRUE@@am__include@ @am__quote@utils/error/$(DEPDIR)/elog.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@utils/mmgr/$(DEPDIR)/aset.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@utils/mmgr/$(DEPDIR)/mcxt.Po@am__quote@
-@AMDEP_TRUE@@am__include@ @am__quote@utils/pcp/$(DEPDIR)/pcp_error.Po@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@utils/pcp/$(DEPDIR)/pcp_stream.Po@am__quote@
-@AMDEP_TRUE@@am__include@ @am__quote@utils/pcp/$(DEPDIR)/pcp_timeout.Po@am__quote@
.c.o:
@am__fastdepCC_TRUE@ $(AM_V_CC)depbase=`echo $@ | sed 's|[^/]*$$|$(DEPDIR)/&|;s|\.o$$||'`;\
diff --git a/src/auth/pool_hba.c b/src/auth/pool_hba.c
index 17d9c7203..e98673550 100644
--- a/src/auth/pool_hba.c
+++ b/src/auth/pool_hba.c
@@ -405,11 +405,7 @@ static void close_all_backend_connections(void)
int i;
POOL_CONNECTION_POOL *p = pool_connection_pool;
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
POOL_SETMASK2(&BlockSig, &oldmask);
diff --git a/src/config/pool_config.c b/src/config/pool_config.c
index 4ae3a0dfa..459277a7a 100644
--- a/src/config/pool_config.c
+++ b/src/config/pool_config.c
@@ -1872,7 +1872,6 @@ int pool_init_config(void)
pool_config->socket_dir = DEFAULT_SOCKET_DIR;
pool_config->pcp_socket_dir = DEFAULT_SOCKET_DIR;
pool_config->backend_socket_dir = NULL;
- pool_config->pcp_timeout = 10;
pool_config->num_init_children = 32;
pool_config->listen_backlog_multiplier = 2;
pool_config->max_pool = 4;
@@ -2367,21 +2366,6 @@ int pool_get_config(char *confpath, POOL_CONFIG_CONTEXT context)
}
pool_config->pcp_socket_dir = str;
}
- else if (!strcmp(key, "pcp_timeout") &&
- CHECK_CONTEXT(INIT_CONFIG|RELOAD_CONFIG, context))
- {
- int v = atoi(yytext);
-
- if (token != POOL_INTEGER || v < 0)
- {
- fclose(fd);
- ereport(error_level,
- (errmsg("invalid configuration for key \"%s\"",key),
- errdetail("invalid value:\"%s\" for key:\"%s\". %s must be >= 0",yytext,key, key)));
- return(-1);
- }
- pool_config->pcp_timeout = v;
- }
else if (!strcmp(key, "num_init_children") && CHECK_CONTEXT(INIT_CONFIG, context))
{
int v = atoi(yytext);
diff --git a/src/config/pool_config.l b/src/config/pool_config.l
index 06634e41a..76cc423e1 100644
--- a/src/config/pool_config.l
+++ b/src/config/pool_config.l
@@ -133,7 +133,6 @@ int pool_init_config(void)
pool_config->socket_dir = DEFAULT_SOCKET_DIR;
pool_config->pcp_socket_dir = DEFAULT_SOCKET_DIR;
pool_config->backend_socket_dir = NULL;
- pool_config->pcp_timeout = 10;
pool_config->num_init_children = 32;
pool_config->listen_backlog_multiplier = 2;
pool_config->max_pool = 4;
@@ -628,21 +627,6 @@ int pool_get_config(char *confpath, POOL_CONFIG_CONTEXT context)
}
pool_config->pcp_socket_dir = str;
}
- else if (!strcmp(key, "pcp_timeout") &&
- CHECK_CONTEXT(INIT_CONFIG|RELOAD_CONFIG, context))
- {
- int v = atoi(yytext);
-
- if (token != POOL_INTEGER || v < 0)
- {
- fclose(fd);
- ereport(error_level,
- (errmsg("invalid configuration for key \"%s\"",key),
- errdetail("invalid value:\"%s\" for key:\"%s\". %s must be >= 0",yytext,key, key)));
- return(-1);
- }
- pool_config->pcp_timeout = v;
- }
else if (!strcmp(key, "num_init_children") && CHECK_CONTEXT(INIT_CONFIG, context))
{
int v = atoi(yytext);
diff --git a/src/include/pcp/libpcp_ext.h b/src/include/pcp/libpcp_ext.h
index 3a9b26490..a784e6c0c 100644
--- a/src/include/pcp/libpcp_ext.h
+++ b/src/include/pcp/libpcp_ext.h
@@ -262,7 +262,6 @@ typedef struct PCPConnInfo{
ConnStateType connState;
PCPResultInfo *pcpResInfo;
FILE *Pfdebug; /* File pointer to output debug infor */
- int errorCode;
}PCPConnInfo;
struct WdInfo;
@@ -286,7 +285,6 @@ extern PCPResultInfo *pcp_recovery_node(PCPConnInfo* pcpConn, int nid);
extern PCPResultInfo *pcp_promote_node(PCPConnInfo* pcpConn, int nid);
extern PCPResultInfo *pcp_promote_node_gracefully(PCPConnInfo* pcpConn,int nid);
extern PCPResultInfo *pcp_watchdog_info(PCPConnInfo* pcpConn, int nid);
-extern void pcp_set_timeout(long sec);
extern PCPResultInfo *pcp_set_backend_parameter(PCPConnInfo* pcpConn,char* parameter_name, char* value);
@@ -303,7 +301,5 @@ extern char *pcp_get_last_error(PCPConnInfo* pcpConn);
* pcp_error.c
* ------------------------------
*/
-//extern ErrorCode errorcode;
-//extern void pcp_errorstr(ErrorCode e);
#endif /* LIBPCP_EXT_H */
diff --git a/src/include/pcp/pcp.h b/src/include/pcp/pcp.h
index 364f12c06..8c0a69e19 100644
--- a/src/include/pcp/pcp.h
+++ b/src/include/pcp/pcp.h
@@ -4,7 +4,7 @@
* pgpool: a language independent connection pool server for PostgreSQL
* written by Tatsuo Ishii
*
- * Copyright (c) 2003-2008 PgPool Global Development Group
+ * Copyright (c) 2003-2015 PgPool Global Development Group
*
* Permission to use, copy, modify, and distribute this software and
* its documentation for any purpose and without fee is hereby
@@ -29,33 +29,14 @@
#define MAX_USER_PASSWD_LEN 128
-typedef enum {
- UNKNOWNERR = 1, /* shouldn't happen */
- EOFERR, /* EOF read by read() */
- NOMEMERR, /* could not allocate memory */
- READERR, /* read() error */
- WRITEERR, /* flush() error */
- TIMEOUTERR, /* select() timeout */
- INVALERR, /* invalid command-line argument(s) number, length, range, etc. */
- CONNERR, /* thrown by connect() */
- NOCONNERR, /* not connected to server */
- SOCKERR, /* thrown by socket() or setsockopt() */
- HOSTERR, /* thrown by gethostbyname() */
- BACKENDERR, /* server dependent error */
- AUTHERR /* authorization failure */
-} ErrorCode;
-
/* --------------------------------
* pcp.c
* --------------------------------
*/
-extern struct timeval pcp_timeout;
/* ------------------------------
* pcp_error.c
* ------------------------------
*/
-extern ErrorCode errorcode;
-extern void pcp_errorstr(ErrorCode e);
#endif /* PCP_H */
diff --git a/src/include/pool.h b/src/include/pool.h
index f5c925624..a936d1963 100644
--- a/src/include/pool.h
+++ b/src/include/pool.h
@@ -360,11 +360,12 @@ extern int my_master_node_id;
#define NO_LOAD_BALANCE "/*NO LOAD BALANCE*/"
#define NO_LOAD_BALANCE_COMMENT_SZ (sizeof(NO_LOAD_BALANCE)-1)
-#define MAX_NUM_SEMAPHORES 4
-#define CONN_COUNTER_SEM 0
-#define REQUEST_INFO_SEM 1
-#define SHM_CACHE_SEM 2
+#define MAX_NUM_SEMAPHORES 5
+#define CONN_COUNTER_SEM 0
+#define REQUEST_INFO_SEM 1
+#define SHM_CACHE_SEM 2
#define QUERY_CACHE_STATS_SEM 3
+#define PCP_REQUEST_SEM 4
#define MAX_REQUEST_QUEUE_SIZE 10
/*
@@ -440,7 +441,6 @@ typedef enum {
* global variables
*/
extern pid_t mypid; /* parent pid */
-extern bool run_as_pcp_child;
typedef enum
{
@@ -453,7 +453,8 @@ typedef enum
PT_LIFECHECK,
PT_FOLLOWCHILD,
PT_WATCHDOG_UTILITY,
- PT_PCP
+ PT_PCP,
+ PT_PCP_WORKER
} ProcessType;
extern ProcessType processType;
@@ -501,7 +502,7 @@ extern bool register_node_operation_request(POOL_REQUEST_KIND kind, int* node_id
extern char *get_config_file_name(void);
extern char *get_hba_file_name(void);
extern void do_child(int *fds);
-extern void pcp_do_child(int unix_fd, int inet_fd, char *pcp_conf_file);
+extern void pcp_main(int unix_fd, int inet_fd);
extern int select_load_balancing_node(void);
extern int pool_init_cp(void);
extern POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
@@ -703,9 +704,15 @@ extern int PgpoolMain(bool discard_status, bool clear_memcache_oidmaps);
/* pcp_child.c */
extern int send_to_pcp_frontend(char* data, int len, bool flush);
extern int pcp_frontend_exists(void);
+extern void pcp_worker_main(int port);
+extern void pcp_mark_recovery_finished(void);
+extern bool pcp_mark_recovery_in_progress(void);
+
/* pgpool_main.c */
extern int pool_send_to_frontend(char* data, int len, bool flush);
extern int pool_frontend_exists(void);
+extern pid_t pool_waitpid(int *status);
+
#endif /* POOL_H */
diff --git a/src/include/pool_config.h b/src/include/pool_config.h
index 94a77ce2e..676ba9fb7 100644
--- a/src/include/pool_config.h
+++ b/src/include/pool_config.h
@@ -75,7 +75,6 @@ typedef struct {
int pcp_port; /* PCP port # to bind */
char *socket_dir; /* pgpool socket directory */
char *pcp_socket_dir; /* PCP socket directory */
- int pcp_timeout; /* PCP timeout for an idle client */
int num_init_children; /* # of children initially pre-forked */
int listen_backlog_multiplier; /* determines the size of the connection queue */
int child_life_time; /* if idle for this seconds, child exits */
diff --git a/src/include/utils/pool_signal.h b/src/include/utils/pool_signal.h
index 53e114d80..593ffb16d 100644
--- a/src/include/utils/pool_signal.h
+++ b/src/include/utils/pool_signal.h
@@ -33,16 +33,21 @@
#include <signal.h>
#ifdef HAVE_SIGPROCMASK
-extern sigset_t UnBlockSig,
- BlockSig,
- AuthBlockSig;
+#define pool_sigset_t sigset_t
+#else
+#define pool_sigset_t int
+#endif
+
+extern pool_sigset_t UnBlockSig,
+BlockSig,
+AuthBlockSig;
+
+
+#ifdef HAVE_SIGPROCMASK
#define POOL_SETMASK(mask) sigprocmask(SIG_SETMASK, mask, NULL)
#define POOL_SETMASK2(mask, oldmask) sigprocmask(SIG_SETMASK, mask, oldmask)
#else
-extern int UnBlockSig,
- BlockSig,
- AuthBlockSig;
#ifndef WIN32
#define POOL_SETMASK(mask) sigsetmask(*((int*)(mask)))
diff --git a/src/libs/pcp/Makefile.am b/src/libs/pcp/Makefile.am
index b37b27058..b1724e9b5 100644
--- a/src/libs/pcp/Makefile.am
+++ b/src/libs/pcp/Makefile.am
@@ -3,8 +3,6 @@ AM_CPPFLAGS = -D_GNU_SOURCE -DPOOL_PRIVATE -I @PGSQL_INCLUDE_DIR@
lib_LTLIBRARIES = libpcp.la
libpcp_la_SOURCES = pcp.c \
pcp_stream.c \
- pcp_error.c \
- pcp_timeout.c \
md5.c \
strlcpy.c
diff --git a/src/libs/pcp/Makefile.in b/src/libs/pcp/Makefile.in
index d1b1f4c66..83479204b 100644
--- a/src/libs/pcp/Makefile.in
+++ b/src/libs/pcp/Makefile.in
@@ -127,8 +127,7 @@ am__uninstall_files_from_dir = { \
am__installdirs = "$(DESTDIR)$(libdir)" "$(DESTDIR)$(includedir)"
LTLIBRARIES = $(lib_LTLIBRARIES)
libpcp_la_LIBADD =
-am_libpcp_la_OBJECTS = pcp.lo pcp_stream.lo pcp_error.lo \
- pcp_timeout.lo md5.lo strlcpy.lo
+am_libpcp_la_OBJECTS = pcp.lo pcp_stream.lo md5.lo strlcpy.lo
libpcp_la_OBJECTS = $(am_libpcp_la_OBJECTS)
AM_V_lt = $(am__v_lt_@AM_V@)
am__v_lt_ = $(am__v_lt_@AM_DEFAULT_V@)
@@ -327,8 +326,6 @@ AM_CPPFLAGS = -D_GNU_SOURCE -DPOOL_PRIVATE -I @PGSQL_INCLUDE_DIR@
lib_LTLIBRARIES = libpcp.la
libpcp_la_SOURCES = pcp.c \
pcp_stream.c \
- pcp_error.c \
- pcp_timeout.c \
md5.c \
strlcpy.c
@@ -418,9 +415,7 @@ distclean-compile:
@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/md5.Plo@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/pcp.Plo@am__quote@
-@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/pcp_error.Plo@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/pcp_stream.Plo@am__quote@
-@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/pcp_timeout.Plo@am__quote@
@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/strlcpy.Plo@am__quote@
.c.o:
diff --git a/src/libs/pcp/pcp.c b/src/libs/pcp/pcp.c
index 1e8ed2a86..24543bcea 100644
--- a/src/libs/pcp/pcp.c
+++ b/src/libs/pcp/pcp.c
@@ -109,7 +109,6 @@ pcp_connect(char *hostname, int port, char *username, char *password, FILE *Pfde
{
pcp_internal_error(pcpConn,
"ERROR: failed to create UNIX domain socket. socket error \"%s\"",strerror(errno));
- pcpConn->errorCode = SOCKERR;
pcpConn->connState = PCP_CONNECTION_BAD;
return pcpConn;
@@ -136,7 +135,6 @@ pcp_connect(char *hostname, int port, char *username, char *password, FILE *Pfde
pcp_internal_error(pcpConn,
"ERROR: connection to socket \"%s\" failed with error \"%s\"",unix_addr.sun_path,strerror(errno));
- pcpConn->errorCode = CONNERR;
pcpConn->connState = PCP_CONNECTION_BAD;
return pcpConn;
}
@@ -148,7 +146,6 @@ pcp_connect(char *hostname, int port, char *username, char *password, FILE *Pfde
{
pcp_internal_error(pcpConn,
"ERROR: failed to create INET domain socket with error \"%s\"",strerror(errno));
- pcpConn->errorCode = SOCKERR;
pcpConn->connState = PCP_CONNECTION_BAD;
return pcpConn;
}
@@ -160,7 +157,6 @@ pcp_connect(char *hostname, int port, char *username, char *password, FILE *Pfde
pcp_internal_error(pcpConn,
"ERROR: set socket option failed with error \"%s\"",strerror(errno));
- pcpConn->errorCode = SOCKERR;
pcpConn->connState = PCP_CONNECTION_BAD;
return pcpConn;
}
@@ -173,7 +169,6 @@ pcp_connect(char *hostname, int port, char *username, char *password, FILE *Pfde
close(fd);
pcp_internal_error(pcpConn,
"ERROR: could not retrieve hostname. gethostbyname failed with error \"%s\"",strerror(errno));
- pcpConn->errorCode = HOSTERR;
pcpConn->connState = PCP_CONNECTION_BAD;
return pcpConn;
@@ -189,7 +184,6 @@ pcp_connect(char *hostname, int port, char *username, char *password, FILE *Pfde
close(fd);
pcp_internal_error(pcpConn,
"ERROR: connection to host \"%s\" failed with error \"%s\"",hostname,strerror(errno));
- pcpConn->errorCode = CONNERR;
pcpConn->connState = PCP_CONNECTION_BAD;
return pcpConn;
}
@@ -201,7 +195,6 @@ pcp_connect(char *hostname, int port, char *username, char *password, FILE *Pfde
close(fd);
pcp_internal_error(pcpConn,
"ERROR: failed to allocate memory");
- pcpConn->errorCode = NOMEMERR;
pcpConn->connState = PCP_CONNECTION_BAD;
return pcpConn;
}
@@ -359,8 +352,6 @@ static PCPResultInfo* process_pcp_response(PCPConnInfo* pcpConn, char sentMsg)
pcp_internal_error(pcpConn,
"ERROR: authentication failed. reason=\"%s\"\n", buf);
setResultStatus(pcpConn, PCP_RES_BACKEND_ERROR);
-
- pcpConn->errorCode = AUTHERR;
}
}
break;
diff --git a/src/main/pgpool_main.c b/src/main/pgpool_main.c
index 62b94041f..f8b6f5e55 100644
--- a/src/main/pgpool_main.c
+++ b/src/main/pgpool_main.c
@@ -128,7 +128,6 @@ static int find_primary_node(void);
static int find_primary_node_repeatedly(void);
static void terminate_all_childrens();
static void system_will_go_down(int code, Datum arg);
-static pid_t pool_waitpid(int *status);
static char* process_name_from_pid(pid_t pid);
static struct sockaddr_un un_addr; /* unix domain socket path */
@@ -553,11 +552,7 @@ process_backend_health_check_failure(int health_check_node_id, int retrycnt)
bool register_node_operation_request(POOL_REQUEST_KIND kind, int* node_id_set, int count)
{
bool failover_in_progress;
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
/*
* if the queue is already full
@@ -627,8 +622,7 @@ pid_t pcp_fork_a_child(int unix_fd, int inet_fd, char *pcp_conf_file)
POOL_SETMASK(&UnBlockSig);
health_check_timer_expired = 0;
reload_config_request = 0;
- run_as_pcp_child = true;
- pcp_do_child(unix_fd, inet_fd, pcp_conf_file);
+ pcp_main(unix_fd, inet_fd);
}
else if (pid == -1)
{
@@ -674,7 +668,6 @@ pid_t fork_a_child(int *fds, int id)
health_check_timer_expired = 0;
reload_config_request = 0;
my_proc_id = id;
- run_as_pcp_child = false;
do_child(fds);
}
else if (pid == -1)
@@ -1961,7 +1954,7 @@ static RETSIGTYPE reap_handler(int sig)
* nothing more than a wrapper function over NOHANG mode waitpid() or wait3()
* depending on the existance of waitpid in the system
*/
-static pid_t pool_waitpid(int *status)
+pid_t pool_waitpid(int *status)
{
#ifdef HAVE_WAITPID
return waitpid(-1, status, WNOHANG);
@@ -2998,7 +2991,7 @@ static void system_will_go_down(int code, Datum arg)
int pool_send_to_frontend(char* data, int len, bool flush)
{
- if (processType == PT_PCP)
+ if (processType == PT_PCP_WORKER)
return send_to_pcp_frontend(data, len, flush);
else if (processType == PT_CHILD)
return send_to_pg_frontend(data, len, flush);
@@ -3007,7 +3000,7 @@ int pool_send_to_frontend(char* data, int len, bool flush)
int pool_frontend_exists(void)
{
- if (processType == PT_PCP)
+ if (processType == PT_PCP_WORKER)
return pcp_frontend_exists();
else if (processType == PT_CHILD)
return pg_frontend_exists();
diff --git a/src/main/pool_globals.c b/src/main/pool_globals.c
index 208e0e0da..bcf06a6ee 100644
--- a/src/main/pool_globals.c
+++ b/src/main/pool_globals.c
@@ -22,8 +22,6 @@
*/
#include "pool.h"
-int debug = 0; /* non 0 if debug option is given (-d). pcp only */
pid_t mypid; /* pgpool parent process id */
-bool run_as_pcp_child;
ProcessType processType;
ProcessState processState;
diff --git a/src/pcp_con/pcp_child.c b/src/pcp_con/pcp_child.c
index a3ac7e82d..13a2d1225 100644
--- a/src/pcp_con/pcp_child.c
+++ b/src/pcp_con/pcp_child.c
@@ -5,7 +5,7 @@
* pgpool: a language independent connection pool server for PostgreSQL
* written by Tatsuo Ishii
*
- * Copyright (c) 2003-2011 PgPool Global Development Group
+ * Copyright (c) 2003-2015 PgPool Global Development Group
*
* Permission to use, copy, modify, and distribute this software and
* its documentation for any purpose and without fee is hereby
@@ -28,6 +28,8 @@
#include <sys/types.h>
#include <sys/socket.h>
+#include <sys/time.h>
+#include <sys/wait.h>
#include <netinet/in.h>
#include <sys/un.h>
#include <arpa/inet.h>
@@ -46,92 +48,58 @@
#include <string.h>
#include <unistd.h>
#include <stdlib.h>
-#include <sys/time.h>
+
#ifdef HAVE_FCNTL_H
#include <fcntl.h>
#endif
#include "pool.h"
-#include "pcp/pcp_stream.h"
-#include "pcp/pcp.h"
-#include "auth/md5.h"
#include "pool_config.h"
-#include "context/pool_process_context.h"
-#include "utils/pool_process_reporting.h"
-#include "watchdog/wd_ext.h"
#include "utils/elog.h"
+#include "parser/pg_list.h"
-#define MAX_FILE_LINE_LEN 512
-#define MAX_USER_PASSWD_LEN 128
-
-extern void pcp_set_timeout(long sec);
-volatile sig_atomic_t pcp_exit_request; /* non 0 means SIGTERM(smart shutdown) or SIGINT(fast shutdown) has arrived */
-
-static RETSIGTYPE die(int sig);
-static PCP_CONNECTION *pcp_do_accept(int unix_fd, int inet_fd);
-static void unset_nonblock(int fd);
-static int user_authenticate(char *buf, char *passwd_file, char *salt, int salt_len);
-static RETSIGTYPE wakeup_handler(int sig);
-static RETSIGTYPE reload_config_handler(int sig);
-static RETSIGTYPE restart_handler(int sig);
-static int pool_detach_node(int node_id, bool gracefully);
-static int pool_promote_node(int node_id, bool gracefully);
-
+static int pcp_unix_fd, pcp_inet_fd;
+volatile bool *pcp_recovery_in_progress;
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;
+List *pcp_worker_children = NULL;
+static volatile sig_atomic_t sigchld_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(PCP_CONNECTION *frontend,char mode);
-static void process_set_configration_parameter(PCP_CONNECTION *frontend,char *buf, int len);
+static RETSIGTYPE pcp_exit_handler(int sig);
+static RETSIGTYPE wakeup_handler_parent(int sig);
+static RETSIGTYPE reload_config_handler(int sig);
+static RETSIGTYPE restart_handler(int sig);
+static RETSIGTYPE reap_handler(int sig);
-static void process_authentication(PCP_CONNECTION *frontend, char *buf, char *pcp_conf_file,
- char* salt, int *random_salt,volatile int *authenticated);
-static void send_md5salt(PCP_CONNECTION *frontend, char* salt);
+static int pcp_do_accept(int unix_fd, int inet_fd);
+static void start_pcp_command_processor_process(int port);
+static void pcp_child_will_die(int code, Datum arg);
+static void pcp_kill_all_children(int sig);
+static void reaper(void);
#define CHECK_RESTART_REQUEST \
do { \
- if (pcp_restart_request) \
+ if (sigchld_request) \
{ \
- ereport(LOG,(errmsg("restart request received in pcp child process"))); \
- exit(1); \
+ reaper(); \
} \
- else \
+ if (pcp_restart_request) \
{ \
- pool_initialize_private_backend_status(); \
+ ereport(LOG,(errmsg("restart request received in pcp child process"))); \
+ pcp_exit_handler(SIGTERM); \
} \
} while (0)
-PCP_CONNECTION* volatile pcp_frontend = NULL;
-void
-pcp_do_child(int unix_fd, int inet_fd, char *pcp_conf_file)
+/*
+ * main entry point for pcp child process
+ */
+void pcp_main(int unix_fd, int inet_fd)
{
sigjmp_buf local_sigjmp_buf;
- MemoryContext PCPMemoryContext;
-
-// PCP_CONNECTION* volatile frontend = NULL;
- volatile int authenticated = 0;
struct timeval uptime;
- char salt[4];
- int random_salt = 0;
- char tos;
- int rsize;
- char *buf = NULL;
ereport(DEBUG1,
(errmsg("I am PCP child with pid:%d",getpid())));
@@ -140,35 +108,29 @@ pcp_do_child(int unix_fd, int inet_fd, char *pcp_conf_file)
init_ps_display("", "", "", "");
gettimeofday(&uptime, NULL);
- srandom((unsigned int) (getpid() ^ uptime.tv_usec));
+ pcp_unix_fd = unix_fd;
+ pcp_inet_fd = inet_fd;
- /* set pcp_read() timeout */
- pcp_set_timeout(pool_config->pcp_timeout);
+ pcp_recovery_in_progress = pool_shared_memory_create(sizeof(bool));
+ *pcp_recovery_in_progress = false;
+ /*
+ * install the call back for preparation of exit
+ */
+ on_system_exit(pcp_child_will_die, (Datum)NULL);
/* set up signal handlers */
- signal(SIGTERM, die);
- signal(SIGINT, die);
- signal(SIGHUP, reload_config_handler);
- signal(SIGQUIT, die);
- signal(SIGCHLD, SIG_DFL);
- signal(SIGUSR1, restart_handler);
- 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);
+ pool_signal(SIGTERM, pcp_exit_handler);
+ pool_signal(SIGINT, pcp_exit_handler);
+ pool_signal(SIGQUIT, pcp_exit_handler);
+ pool_signal(SIGHUP, reload_config_handler);
+ pool_signal(SIGCHLD, reap_handler);
- /* Initialize my backend status */
- pool_initialize_private_backend_status();
+ pool_signal(SIGUSR1, restart_handler);
+ pool_signal(SIGUSR2, wakeup_handler_parent);
+ pool_signal(SIGPIPE, SIG_IGN);
+ pool_signal(SIGALRM, SIG_IGN);
- /* Initialize process context */
- pool_init_process_context();
+ MemoryContextSwitchTo(TopMemoryContext);
if (sigsetjmp(local_sigjmp_buf, 1) != 0)
{
@@ -176,247 +138,30 @@ pcp_do_child(int unix_fd, int inet_fd, char *pcp_conf_file)
EmitErrorReport();
pool_signal(SIGALRM, SIG_IGN);
- if(pcp_frontend)
- pcp_close(pcp_frontend);
-
- pcp_frontend = NULL;
- authenticated = 0;
MemoryContextSwitchTo(TopMemoryContext);
FlushErrorState();
}
/* We can now handle ereport(ERROR) */
PG_exception_stack = &local_sigjmp_buf;
-
for(;;)
{
- MemoryContextSwitchTo(PCPMemoryContext);
- MemoryContextResetAndDeleteChildren(PCPMemoryContext);
-
+ int port;
errno = 0;
-
CHECK_RESTART_REQUEST;
- if (pcp_frontend == NULL)
- {
- pcp_frontend = pcp_do_accept(unix_fd, inet_fd);
- if (pcp_frontend == NULL)
- continue;
-
- unset_nonblock(pcp_frontend->fd);
- }
-
- /* read a PCP packet */
- if (pcp_read(pcp_frontend, &tos, 1))
- {
- if (errorcode == TIMEOUTERR)
- {
- ereport(ERROR,
- (errmsg("unable to read from client"),
- errdetail("pcp_read time out")));
- }
- ereport(FATAL,
- (errmsg("unable to read from client"),
- errdetail("pcp_read failed with error : \"%s\"",strerror(errno))));
-
- }
- if (pcp_read(pcp_frontend, &rsize, sizeof(int)))
- {
- if (errorcode == TIMEOUTERR)
- {
- ereport(ERROR,
- (errmsg("unable to read from client"),
- errdetail("pcp_read time out")));
- }
- 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 *)palloc(rsize - sizeof(int));
- if (pcp_read(pcp_frontend, buf, rsize - sizeof(int)))
- {
- if (errorcode == TIMEOUTERR)
- {
- ereport(ERROR,
- (errmsg("unable to read from client"),
- errdetail("pcp_read time out")));
- }
- 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'))
- {
- ereport(ERROR,
- (errmsg("authentication failed for new PCP connection"),
- errdetail("connection not authorized")));
- }
-
- /* process a request */
- ereport(DEBUG1,
- (errmsg("received PCP packet"),
- errdetail("PCP packet type of service '%c'", tos)));
-
- set_ps_display("PCP: processing a request", false);
-
- if (tos == 'C' || tos == 'd' || tos == 'D' || tos == 'j' ||
- tos == 'J' || tos == 'O' || tos == 'T')
- {
- if (Req_info->switching)
- {
- if(Req_info->request_queue_tail != Req_info->request_queue_head)
- {
- POOL_REQUEST_KIND reqkind;
- reqkind = Req_info->request[(Req_info->request_queue_head +1) % MAX_REQUEST_QUEUE_SIZE].kind;
-
- if (reqkind == NODE_UP_REQUEST)
- ereport(ERROR,
- (errmsg("failed to process PCP request at the moment"),
- errdetail("failback is in progress")));
- else if (reqkind == NODE_DOWN_REQUEST)
- ereport(ERROR,
- (errmsg("failed to process PCP request at the moment"),
- errdetail("failover is in progress")));
- else if (reqkind == PROMOTE_NODE_REQUEST)
- ereport(ERROR,
- (errmsg("failed to process PCP request at the moment"),
- errdetail("promote node operation is in progress")));
- ereport(ERROR,
- (errmsg("failed to process PCP request at the moment"),
- errdetail("operation is in progress")));
- }
- }
- }
-
- switch (tos)
+ port = pcp_do_accept(unix_fd, inet_fd);
+ if(port > 0)
{
- case 'R': /* authentication */
- process_authentication(pcp_frontend, buf, pcp_conf_file,
- salt, &random_salt, &authenticated);
- break;
-
- case 'M': /* md5 salt */
- send_md5salt(pcp_frontend, salt);
- random_salt = 1;
- break;
- case 'A': /* set configuration parameter */
- process_set_configration_parameter(pcp_frontend,buf,rsize);
- break;
- case 'L': /* node count */
- inform_node_count(pcp_frontend);
- break;
- case 'I': /* node info */
- inform_node_info(pcp_frontend, buf);
- break;
- case 'N': /* process count */
- inform_process_count(pcp_frontend);
- break;
- case 'P': /* process info */
- inform_process_info(pcp_frontend, buf);
- break;
- case 'S': /* SystemDB info */
- inform_systemDB_info(pcp_frontend);
- break;
- case 'W': /* watchdog info */
- inform_watchdog_info(pcp_frontend, buf);
- break;
- case 'D': /* detach node */
- case 'd': /* detach node gracefully */
- process_detach_node(pcp_frontend, buf, tos);
- break;
- case 'C': /* attach node */
- process_attach_node(pcp_frontend, buf);
- break;
- case 'T':
- process_shutown_request(pcp_frontend, buf[0]);
- break;
-
- case 'O': /* recovery request */
- process_recovery_request(pcp_frontend, buf);
- break;
-
- case 'B': /* status request*/
- process_status_request(pcp_frontend);
- break;
-
- case 'J': /* promote node */
- case 'j': /* promote node gracefully */
- process_promote_node(pcp_frontend,buf,tos);
- break;
-
- case 'F':
- ereport(DEBUG1,
- (errmsg("PCP processing request, stop online recovery")));
- break;
-
- case 'X': /* disconnect */
- ereport(DEBUG1,
- (errmsg("PCP processing request, client disconnecting"),
- errdetail("closing PCP connection")));
- authenticated = 0;
- pcp_close(pcp_frontend);
- pcp_frontend = NULL;
- break;
-
- default:
- ereport(FATAL,
- (errmsg("PCP processing request"),
- errdetail("unknown PCP packet type \"%c\"",tos)));
+ start_pcp_command_processor_process(port);
}
- /* seems ok. cancel idle check timer */
- pool_signal(SIGALRM, SIG_IGN);
}
-
- exit(0);
-}
-
-static RETSIGTYPE
-die(int sig)
-{
- pcp_exit_request = 1;
- ereport(DEBUG1,
- (errmsg("PCP child receives shutdown request signal %d", sig)));
-
- switch (sig)
- {
- case SIGTERM: /* smart shutdown */
- case SIGINT: /* fast shutdown */
- case SIGQUIT: /* immediate shutdown */
- exit(0);
- break;
- default:
- break;
- }
-
- exit(1);
-}
-
-static RETSIGTYPE
-wakeup_handler(int sig)
-{
- pcp_wakeup_request = 1;
}
-static RETSIGTYPE
-restart_handler(int sig)
-{
- pcp_restart_request = 1;
-}
-
-static PCP_CONNECTION *
+static int
pcp_do_accept(int unix_fd, int inet_fd)
{
- PCP_CONNECTION *pc = NULL;
-
fd_set readmask;
int fds;
-
struct sockaddr addr;
socklen_t addrlen;
int fd = 0;
@@ -431,21 +176,18 @@ pcp_do_accept(int unix_fd, int inet_fd)
FD_SET(inet_fd, &readmask);
fds = select(Max(unix_fd, inet_fd)+1, &readmask, NULL, NULL, NULL);
-
if (fds == -1)
{
if (errno == EAGAIN || errno == EINTR)
- return NULL;
+ return -1;
ereport(ERROR,
- (errmsg("unable to accept new pcp connection"),
- errdetail("select system call failed with error : \"%s\"",strerror(errno))));
+ (errmsg("unable to accept new pcp connection"),
+ errdetail("select system call failed with error : \"%s\"",strerror(errno))));
}
-
if (FD_ISSET(unix_fd, &readmask))
{
fd = unix_fd;
}
-
if (FD_ISSET(inet_fd, &readmask))
{
fd = inet_fd;
@@ -463,15 +205,14 @@ pcp_do_accept(int unix_fd, int inet_fd)
*/
if (errno != EAGAIN && errno != EWOULDBLOCK)
ereport(ERROR,
- (errmsg("unable to accept new pcp connection"),
- errdetail("socket accept system call failed with error : \"%s\"",strerror(errno))));
+ (errmsg("unable to accept new pcp connection"),
+ errdetail("socket accept system call failed with error : \"%s\"",strerror(errno))));
}
-
if (pcp_got_sighup)
{
- MemoryContext oldContext = MemoryContextSwitchTo(TopMemoryContext);
+ MemoryContext oldContext = MemoryContextSwitchTo(TopMemoryContext);
pool_get_config(get_config_file_name(), RELOAD_CONFIG);
- MemoryContextSwitchTo(oldContext);
+ MemoryContextSwitchTo(oldContext);
pcp_got_sighup = 0;
}
ereport(DEBUG2,
@@ -479,15 +220,14 @@ pcp_do_accept(int unix_fd, int inet_fd)
if (inet)
{
int on = 1;
-
if (setsockopt(afd, IPPROTO_TCP, TCP_NODELAY,
(char *) &on,
sizeof(on)) < 0)
{
close(afd);
ereport(ERROR,
- (errmsg("unable to accept new pcp connection"),
- errdetail("setsockopt system call failed with error : \"%s\"",strerror(errno))));
+ (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,
@@ -495,1045 +235,253 @@ pcp_do_accept(int unix_fd, int inet_fd)
{
close(afd);
ereport(ERROR,
- (errmsg("unable to accept new pcp connection"),
- errdetail("setsockopt system call failed with error : \"%s\"",strerror(errno))));
+ (errmsg("unable to accept new pcp connection"),
+ errdetail("setsockopt system call failed with error : \"%s\"",strerror(errno))));
}
}
- pc = pcp_open(afd);
- return pc;
+ return afd;
}
/*
- * unset non-block flag
+ * forks a new pcp worker child
*/
-static void
-unset_nonblock(int fd)
+static void start_pcp_command_processor_process(int port)
{
- int var;
-
- /* set fd to non-blocking */
- var = fcntl(fd, F_GETFL, 0);
- if (var == -1)
+ pid_t pid = fork();
+ if (pid == 0)/* child */
+ {
+ /* Set the process type variable */
+ processType = PT_PCP_WORKER;
+ on_exit_reset();
+ /* Close the listen sockets sockets */
+ close(pcp_unix_fd);
+ close(pcp_inet_fd);
+ /* call PCP child main */
+ if (pcp_worker_children)
+ list_free(pcp_worker_children);
+ pcp_worker_children= NULL;
+ POOL_SETMASK(&UnBlockSig);
+ pcp_worker_main(port); /* Never returns */
+ }
+ else if (pid == -1)
{
ereport(FATAL,
- (errmsg("unable to connect"),
- errdetail("fcntl system call failed with error : \"%s\"",strerror(errno))));
-
+ (errmsg("fork() failed. reason: %s", strerror(errno))));
}
- if (fcntl(fd, F_SETFL, var & ~O_NONBLOCK) == -1)
+ else /* parent */
{
- ereport(FATAL,
- (errmsg("unable to connect"),
- errdetail("fcntl system call failed with error : \"%s\"",strerror(errno))));
+ ereport(LOG,
+ (errmsg("forked new pcp worker, pid=%d socket=%d",
+ (int) pid, (int) port)));
+ /* close the port in parent process. It is only consumed by child */
+ close(port);
+ /* Add it to the list */
+ pcp_worker_children = lappend_int(pcp_worker_children, (int)pid);
}
}
/*
- * see if received username and password matches with one in the file
+ * sends the signal to all children of pcp child process
*/
-static int
-user_authenticate(char *buf, char *passwd_file, char *salt, int salt_len)
-{
- FILE *fp = NULL;
- char packet_username[MAX_USER_PASSWD_LEN+1];
- char packet_password[MAX_USER_PASSWD_LEN+1];
- char encrypt_buf[(MD5_PASSWD_LEN+1)*2];
- char file_username[MAX_USER_PASSWD_LEN+1];
- char file_password[MAX_USER_PASSWD_LEN+1];
- char *index = NULL;
- static char line[MAX_FILE_LINE_LEN+1];
- int i, len;
-
- /* strcpy() should be OK, but use strncpy() to be extra careful */
- strncpy(packet_username, buf, MAX_USER_PASSWD_LEN);
- index = (char *) memchr(buf, '\0', MAX_USER_PASSWD_LEN);
- if (index == NULL)
- {
- ereport(ERROR,
- (errmsg("failed to authenticate PCP user"),
- errdetail("error while reading authentication packet")));
- return 0;
- }
- strncpy(packet_password, ++index, MAX_USER_PASSWD_LEN);
-
- fp = fopen(passwd_file, "r");
- if (fp == NULL)
- {
- ereport(ERROR,
- (errmsg("failed to authenticate PCP user"),
- errdetail("could not open %s. reason: %s", passwd_file, strerror(errno))));
- return 0;
- }
-
- /* for now, I don't care if duplicate username exists in the config file */
- while ((fgets(line, MAX_FILE_LINE_LEN, fp)) != NULL)
- {
- i = 0;
- len = 0;
-
- if (line[0] == '\n')
- continue;
- if (line[0] == '#')
- continue;
-
- while (line[i] != ':')
- {
- len++;
- if (++i > MAX_USER_PASSWD_LEN)
- {
- fclose(fp);
- ereport(ERROR,
- (errmsg("failed to authenticate PCP user"),
- errdetail("username read from file \"%s\" is larger than maximum allowed username length [%d]", passwd_file, MAX_USER_PASSWD_LEN)));
- return 0;
- }
- }
- memcpy(file_username, line, len);
- file_username[len] = '\0';
-
- if (strcmp(packet_username, file_username) != 0)
- continue;
-
- i++;
- len = 0;
- while (line[i] != '\n' && line[i] != '\0')
- {
- len++;
- if (++i > MAX_USER_PASSWD_LEN)
- {
- fclose(fp);
- ereport(ERROR,
- (errmsg("failed to authenticate PCP user"),
- errdetail("password read from file \"%s\" is larger than maximum allowed password length [%d]", passwd_file, MAX_USER_PASSWD_LEN)));
- return 0;
- }
- }
-
- memcpy(file_password, line+strlen(file_username)+1, len);
- file_password[len] = '\0';
-
- pool_md5_encrypt(file_password, file_username, strlen(file_username),
- encrypt_buf + MD5_PASSWD_LEN + 1);
- encrypt_buf[(MD5_PASSWD_LEN+1)*2-1] = '\0';
-
- pool_md5_encrypt(encrypt_buf+MD5_PASSWD_LEN+1, salt, salt_len,
- encrypt_buf);
- encrypt_buf[MD5_PASSWD_LEN] = '\0';
-
- if (strcmp(encrypt_buf, packet_password) == 0)
- {
- fclose(fp);
- return 1;
- }
- }
-
- fclose(fp);
- return 0;
-}
-
-/* SIGHUP handler */
-static RETSIGTYPE reload_config_handler(int sig)
-{
- pcp_got_sighup = 1;
-}
-
-/* Dedatch a node */
-static int pool_detach_node(int node_id, bool gracefully)
+static void
+pcp_kill_all_children(int sig)
{
- if (!gracefully)
- {
- notice_backend_error(node_id); /* send failover request */
- return 0;
- }
-
- /*
- * Wait until all frontends exit
- */
- *InRecovery = RECOVERY_DETACH; /* This wiil ensure that new incoming
- * connection requests are blocked */
-
- if (wait_connection_closed())
+ /* forward wakeup requests to children */
+ ListCell* lc;
+ foreach(lc, pcp_worker_children)
{
- /* wait timed out */
- finish_recovery();
- return -1;
+ pid_t pid = (pid_t)lfirst_int(lc);
+ kill(pid,sig);
}
-
- /*
- * Now all frontends have gone. Let's do failover.
- */
- notice_backend_error(node_id); /* send failover request */
-
- /*
- * Wait for failover completed.
- */
- pcp_wakeup_request = 0;
-
- while (!pcp_wakeup_request)
- {
- struct timeval t = {1, 0};
- select(0, NULL, NULL, NULL, &t);
- }
- pcp_wakeup_request = 0;
-
- /*
- * Start to accept incoming connections and send SIGUSR2 to pgpool
- * parent to distribute SIGUSR2 all pgpool children.
- */
- finish_recovery();
-
- return 0;
}
-
-/* Promote a node */
-static int pool_promote_node(int node_id, bool gracefully)
+/*
+ * handle SIGCHLD
+ */
+static RETSIGTYPE reap_handler(int sig)
{
- if (!gracefully)
- {
- promote_backend(node_id); /* send promote request */
- return 0;
- }
-
- /*
- * Wait until all frontends exit
- */
- *InRecovery = RECOVERY_PROMOTE; /* This wiil ensure that new incoming
- * connection requests are blocked */
-
- if (wait_connection_closed())
- {
- /* wait timed out */
- finish_recovery();
- return -1;
- }
-
- /*
- * Now all frontends have gone. Let's do failover.
- */
- promote_backend(node_id); /* send promote request */
-
- /*
- * Wait for failover completed.
- */
- pcp_wakeup_request = 0;
-
- while (!pcp_wakeup_request)
- {
- struct timeval t = {1, 0};
- select(0, NULL, NULL, NULL, &t);
- }
- pcp_wakeup_request = 0;
-
- /*
- * Start to accept incoming connections and send SIGUSR2 to pgpool
- * parent to distribute SIGUSR2 all pgpool children.
- */
- finish_recovery();
-
- return 0;
+ POOL_SETMASK(&BlockSig);
+ sigchld_request = 1;
+ POOL_SETMASK(&UnBlockSig);
}
-static void
-inform_process_count(PCP_CONNECTION *frontend)
+static void reaper(void)
{
- 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);
+ pid_t pid;
+ int status;
ereport(DEBUG1,
- (errmsg("PCP: informing process count"),
- errdetail("%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;
+ (errmsg("PCP child reaper handler")));
- proc_id = atoi(buf);
+ /* clear SIGCHLD request */
+ sigchld_request = 0;
- if ((proc_id != 0) && (pool_get_process_info(proc_id) == NULL))
- {
- ereport(ERROR,
- (errmsg("informing process info failed"),
- errdetail("invalid process ID : %s",buf)));
- }
- else
+ while ((pid = pool_waitpid(&status)) > 0)
{
- /* 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)
+ if(WIFEXITED(status))
{
- snprintf(con_info_size, sizeof(con_info_size), "%d", num_proc);
+ if(WEXITSTATUS(status) == POOL_EXIT_FATAL)
+ ereport(LOG,
+ (errmsg("PCP worker process with pid: %d exit with FATAL ERROR.", pid)));
+ else if(WEXITSTATUS(status) == POOL_EXIT_SUCCESS)
+ ereport(LOG,
+ (errmsg("PCP process with pid: %d exit with SUCCESS.", pid)));
}
- else
+ if (WIFSIGNALED(status))
{
- snprintf(con_info_size, sizeof(con_info_size), "%d", pool_config->max_pool * NUM_BACKENDS);
+ /* Child terminated by segmentation fault. Report it */
+ if(WTERMSIG(status) == SIGSEGV)
+ ereport(WARNING,
+ (errmsg("PCP process with pid: %d was terminated by segmentation fault",pid)));
+ else
+ ereport(LOG,
+ (errmsg("PCP process with pid: %d exits with status %d by signal %d", pid, status, WTERMSIG(status))));
}
-
- 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);
- ereport(DEBUG1,
- (errmsg("PCP informing process info"),
- errdetail("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)
- {
- ereport(ERROR,
- (errmsg("informing system db info failed"),
- errdetail("system DB is not defined")));
- }
- 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);
-
- ereport(DEBUG1,
- (errmsg("PCP: informing systemDB info"),
- errdetail("retrieved SystemDB information from shared memory")));
+ else
+ ereport(LOG,
+ (errmsg("PCP process with pid: %d exits with status %d",pid, status)));
+ ereport(DEBUG2,
+ (errmsg("going to remove pid: %d from pid list having %d elements",pid, list_length(pcp_worker_children))));
+ /* remove the pid of process from the list */
+ pcp_worker_children = list_delete_int(pcp_worker_children,pid);
+ ereport(DEBUG2,
+ (errmsg("new list have %d elements",list_length(pcp_worker_children))));
}
}
-static void
-inform_watchdog_info(PCP_CONNECTION *frontend,char *buf)
-{
- int wd_index;
- int wsize;
- char code[] = "CommandComplete";
- char pgpool_port_str[6];
- char wd_port_str[6];
- char status[2];
-
- WdInfo *wi = NULL;
-
- if (!pool_config->use_watchdog)
- ereport(ERROR,
- (errmsg("PCP: informing watchdog info failed"),
- errdetail("watcdhog is not enabled")));
-
- wd_index = atoi(buf);
- wi = wd_get_watchdog_info(wd_index);
-
- if (wi == NULL)
- ereport(ERROR,
- (errmsg("PCP: informing watchdog info failed"),
- errdetail("invalid watchdog index")));
-
- ereport(DEBUG2,
- (errmsg("PCP: informing watchdog info"),
- errdetail("retrieved node information from shared memory")));
-
- 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);
-}
-
-static void
-inform_node_info(PCP_CONNECTION *frontend,char *buf)
-{
- int node_id;
- int wsize;
- char code[] = "CommandComplete";
- char port_str[6];
- char status[2];
- char weight_str[20];
-
- BackendInfo *bi = NULL;
-
- node_id = atoi(buf);
- bi = pool_get_node_info(node_id);
-
- if (bi == NULL)
- ereport(ERROR,
- (errmsg("informing node info failed"),
- errdetail("invalid node ID")));
-
- ereport(DEBUG2,
- (errmsg("PCP: informing node info"),
- errdetail("retrieved node information from shared memory")));
-
- 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);
-
-}
-
-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);
-
- ereport(DEBUG1,
- (errmsg("PCP: informing node count"),
- errdetail("%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);
- ereport(DEBUG1,
- (errmsg("PCP: processing detach node"),
- errdetail("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)
+static RETSIGTYPE
+pcp_exit_handler(int sig)
{
- int node_id;
- int wsize;
- char code[] = "CommandComplete";
+ pid_t wpid;
- node_id = atoi(buf);
+ POOL_SETMASK(&AuthBlockSig);
ereport(DEBUG1,
- (errmsg("PCP: processing attach node"),
- errdetail("attaching Node ID %d", node_id)));
-
- send_failback_request(node_id,true);
-
- 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);
-}
-
+ (errmsg("PCP child receives shutdown request signal %d, Forwarding to all children", sig)));
-static void
-process_recovery_request(PCP_CONNECTION *frontend,char *buf)
-{
- int node_id;
- int wsize;
- char code[] = "CommandComplete";
-
- node_id = atoi(buf);
-
- if ( (node_id < 0) || (node_id >= pool_config->backend_desc->num_backends) )
- {
- ereport(ERROR,
- (errmsg("process recovery request failed"),
- errdetail("node id %d is not valid", node_id)));
- }
+ pcp_kill_all_children(sig);
- 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))
+ if (sig == SIGTERM) /* smart shutdown */
{
- if (MASTER_SLAVE && !strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP))
- ereport(ERROR,
- (errmsg("process recovery request failed"),
- errdetail("primary server cannot be recovered by online recovery.")));
- else
- ereport(ERROR,
- (errmsg("process recovery request failed"),
- errdetail("recovery request is accepted only in replication mode or stereaming replication mode.")));
+ ereport(DEBUG1,
+ (errmsg("PCP child receives smart shutdown request")));
+ /* close the listening sockets */
+ close(pcp_unix_fd);
+ close(pcp_inet_fd);
}
- else
+ else if (sig == SIGINT)
{
ereport(DEBUG1,
- (errmsg("PCP: processing recovery request"),
- errdetail("start online recovery")));
- PG_TRY();
- {
- start_recovery(node_id);
- finish_recovery();
- 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);
- }
- PG_CATCH();
- {
- finish_recovery();
- PG_RE_THROW();
-
- }PG_END_TRY();
-
+ (errmsg("PCP child receives fast shutdown request")));
}
- 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++)
+ else if (sig == SIGQUIT)
{
- 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);
+ ereport(DEBUG1,
+ (errmsg("PCP child receives immediate shutdown request")));
}
- 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);
- ereport(DEBUG1,
- (errmsg("PCP: processing status request"),
- errdetail("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) )
- ereport(ERROR,
- (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))
- {
- ereport(FATAL,
- (errmsg("invalid pgpool mode for process recovery request"),
- errdetail("not in streaming replication mode, can't promote node id %d", node_id)));
-
- }
+ POOL_SETMASK(&UnBlockSig);
- if (node_id == PRIMARY_NODE_ID)
+ if (list_length(pcp_worker_children) > 0)
{
- ereport(FATAL,
- (errmsg("invalid pgpool mode for process recovery request"),
- errdetail("specified node is already primary node, can't promote node id %d", node_id)));
-
+ do
+ {
+ wpid = wait(NULL);
+ }while (wpid > 0 || (wpid == -1 && errno == EINTR));
+
+ if (wpid == -1 && errno != ECHILD)
+ ereport(WARNING,
+ (errmsg("wait() on pcp worker children failed. reason:%s", strerror(errno))));
+ list_free(pcp_worker_children);
}
- ereport(DEBUG1,
- (errmsg("PCP: processing promote node"),
- errdetail("promoting Node ID %d", node_id)));
- pool_promote_node(node_id, gracefully);
+ pcp_worker_children = NULL;
- 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);
+ exit(0);
}
-static void
-process_authentication(PCP_CONNECTION *frontend, char *buf, char *pcp_conf_file,
- char* salt, int *random_salt,volatile int *authenticated)
+/* Wakeup signal handler for pcp parent process */
+static RETSIGTYPE
+wakeup_handler_parent(int sig)
{
- int wsize;
-
- if (*random_salt)
- {
- *authenticated = user_authenticate(buf, pcp_conf_file, salt, 4);
- }
- if (!*random_salt || !*authenticated)
- {
- ereport(ERROR,
- (errmsg("authentication failed"),
- errdetail("username and/or password does not match")));
-
- *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));
- do_pcp_flush(frontend);
- *random_salt = 0;
-
- ereport(DEBUG1,
- (errmsg("PCP: processing authentication request"),
- errdetail("authentication OK")));
- }
+ /* forward wakeup signal to all children */
+ pcp_kill_all_children(SIGUSR2);
}
-static void
-send_md5salt(PCP_CONNECTION *frontend, char* salt)
+static RETSIGTYPE
+restart_handler(int sig)
{
- int wsize;
- ereport(DEBUG1,
- (errmsg("PCP: sending md5 salt to client")));
-
- 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);
-
-
+ pcp_restart_request = 1;
}
-static void
-process_shutown_request(PCP_CONNECTION *frontend, char mode)
+/* SIGHUP handler */
+static RETSIGTYPE reload_config_handler(int sig)
{
- char code[] = "CommandComplete";
- pid_t ppid = getppid();
- int sig,len;
-
- if (mode == 's')
- {
- ereport(DEBUG1,
- (errmsg("PCP: processing shutdown request"),
- errdetail("sending SIGTERM to the parent process with PID:%d", ppid)));
- sig = SIGTERM;
- }
- else if (mode == 'f')
- {
- ereport(DEBUG1,
- (errmsg("PCP: processing shutdown request"),
- errdetail("sending SIGINT to the parent process with PID:%d", ppid)));
- sig = SIGINT;
- }
- else if (mode == 'i')
- {
- ereport(DEBUG1,
- (errmsg("PCP: processing shutdown request"),
- errdetail("sending SIGQUIT to the parent process with PID:%d", ppid)));
- sig = SIGQUIT;
- }
- else
- {
- ereport(ERROR,
- (errmsg("PCP: error while processing shutdown request"),
- errdetail("invalid shutdown mode \"%c\"", mode)));
- }
-
- pcp_write(frontend, "t", 1);
- len = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &len, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- do_pcp_flush(frontend);
-
- pool_signal_parent(sig);
-
+ pcp_got_sighup = 1;
}
-static void
-process_set_configration_parameter(PCP_CONNECTION *frontend,char *buf, int len)
+static void pcp_child_will_die(int code, Datum arg)
{
- char* param_name;
- char* param_value;
- int wsize;
- char code[] = "CommandComplete";
+ pid_t wpid;
+ /*
+ * This is supposed to be called from main process
+ */
+ if(processType != PT_PCP)
+ return;
+ if (list_length(pcp_worker_children) <= 0)
+ return;
+ pcp_kill_all_children(SIGINT);
+ /* wait for all children to exit */
+ do
+ {
+ wpid = wait(NULL);
+ }while (wpid > 0 || (wpid == -1 && errno == EINTR));
- param_name = buf;
- if(param_name == NULL)
- ereport(ERROR,
- (errmsg("PCP: set configuration parameter failed"),
- errdetail("invalid pcp packet received from client")));
-
- param_value = (char *) memchr(buf, '\0', len) + 1;
- if(param_value == NULL)
- ereport(ERROR,
- (errmsg("PCP: set configuration parameter failed"),
- errdetail("invalid pcp packet received from client")));
-
- ereport(LOG,
- (errmsg("PCP: set configuration parameter, \"%s TO %s\"",param_name,param_value)));
-
- if(strcasecmp(param_name, "client_min_messages") == 0)
- {
- const char *ordered_valid_values[] = {"debug5","debug4","debug3","debug2","debug1","log","commerror","info","notice","warning","error",NULL};
- bool found = false;
- int i;
- for(i=0; ; i++)
- {
- char* valid_val = (char*)ordered_valid_values[i];
- if(!valid_val)
- break;
-
- if (!strcasecmp(param_value, valid_val))
- {
- found = true;
- pool_config->client_min_messages = i + 10;
- ereport(DEBUG1,
- (errmsg("PCP setting parameter \"%s\" to \"%s\"",param_name,param_value)));
- break;
- }
- }
- if (!found)
- ereport(ERROR,
- (errmsg("PCP: set configuration parameter failed"),
- errdetail("invalid value \"%s\" for parameter \"%s\"",param_value,param_name)));
- }
- else
- ereport(ERROR,
- (errmsg("PCP: set configuration parameter failed"),
- errdetail("invalid parameter \"%s\"",param_name)));
+ if (wpid == -1 && errno != ECHILD)
+ ereport(WARNING,
+ (errmsg("wait() on pcp worker children failed. reason:%s", strerror(errno))));
-
- pcp_write(frontend, "a", 1);
- wsize = htonl(sizeof(code) + sizeof(int));
- pcp_write(frontend, &wsize, sizeof(int));
- pcp_write(frontend, code, sizeof(code));
- do_pcp_flush(frontend);
+ POOL_SETMASK(&UnBlockSig);
}
-/*
- * Wrapper around pcp_flush which throws FATAL error when pcp_flush fails
+
+/*
+ * sets the shared memory flag to indicate pcp recovery command
+ * is in progress. If the flag is already set the function returns
+ * false.
*/
-static void
-do_pcp_flush(PCP_CONNECTION *frontend)
+bool pcp_mark_recovery_in_progress(void)
{
- if (pcp_flush(frontend) < 0)
- ereport(FATAL,
- (errmsg("failed to flush data to client"),
- errdetail("pcp_flush failed with error : \"%s\"",strerror(errno))));
+ bool command_already_inprogress;
+ pool_sigset_t oldmask;
+ /*
+ * only pcp worker is allowd to make this call
+ */
+ if(processType != PT_PCP_WORKER)
+ return false;
+
+ POOL_SETMASK2(&BlockSig, &oldmask);
+ pool_semaphore_lock(PCP_REQUEST_SEM);
+ command_already_inprogress = *pcp_recovery_in_progress;
+ *pcp_recovery_in_progress = true;
+ pool_semaphore_unlock(PCP_REQUEST_SEM);
+ POOL_SETMASK(&oldmask);
+ return (command_already_inprogress == false);
}
-int send_to_pcp_frontend(char* data, int len, bool flush)
+/*
+ * unsets the shared memory flag to indicate pcp recovery command
+ * is finsihed.
+ */
+void pcp_mark_recovery_finished(void)
{
- int ret;
- if (processType != PT_PCP || pcp_frontend == NULL)
- return -1;
- ret = pcp_write(pcp_frontend, data, len);
- if (flush && !ret)
- ret = pcp_flush(pcp_frontend);
- return ret;
+ pool_sigset_t oldmask;
+ /*
+ * only pcp worker is allowd to make this call
+ */
+ if(processType != PT_PCP_WORKER)
+ return;
+
+ POOL_SETMASK2(&BlockSig, &oldmask);
+ pool_semaphore_lock(PCP_REQUEST_SEM);
+ *pcp_recovery_in_progress = false;
+ pool_semaphore_unlock(PCP_REQUEST_SEM);
+ POOL_SETMASK(&oldmask);
}
-int pcp_frontend_exists(void)
-{
- if (processType != PT_PCP || pcp_frontend == NULL)
- return -1;
- return 0;
-}
diff --git a/src/pcp_con/pcp_worker.c b/src/pcp_con/pcp_worker.c
new file mode 100644
index 000000000..7cd1317ef
--- /dev/null
+++ b/src/pcp_con/pcp_worker.c
@@ -0,0 +1,1419 @@
+/* -*-pcp_worker.c-*- */
+/*
+ *
+ * pgpool: a language independent connection pool server for PostgreSQL
+ * written by Tatsuo Ishii
+ *
+ * Copyright (c) 2003-2015 PgPool Global Development Group
+ *
+ * Permission to use, copy, modify, and distribute this software and
+ * its documentation for any purpose and without fee is hereby
+ * granted, provided that the above copyright notice appear in all
+ * copies and that both that copyright notice and this permission
+ * notice appear in supporting documentation, and that the name of the
+ * author not be used in advertising or publicity pertaining to
+ * distribution of the software without specific, written prior
+ * permission. The author makes no representations about the
+ * suitability of this software for any purpose. It is provided "as
+ * is" without express or implied warranty.
+ *
+ * pcp_worker.c: PCP worker child process main
+ *
+ */
+
+#include "config.h"
+#include "pool.h"
+#include "utils/palloc.h"
+#include "utils/memutils.h"
+
+#include <arpa/inet.h>
+#include <signal.h>
+
+#include <stdio.h>
+#include <errno.h>
+#include <string.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <sys/time.h>
+
+#ifdef HAVE_FCNTL_H
+#include <fcntl.h>
+#endif
+
+#include "pool.h"
+#include "pcp/pcp_stream.h"
+#include "pcp/pcp.h"
+#include "auth/md5.h"
+#include "pool_config.h"
+#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
+
+extern char pcp_conf_file[POOLMAXPATHLEN+1]; /* global variable defined in main.c holds the path for pcp.conf */
+volatile sig_atomic_t pcp_worker_wakeup_request = 0;
+PCP_CONNECTION* volatile pcp_frontend = NULL;
+
+static RETSIGTYPE die(int sig);
+static RETSIGTYPE wakeup_handler_child(int sig);
+
+static void unset_nonblock(int fd);
+static int user_authenticate(char *buf, char *passwd_file, char *salt, int salt_len);
+static void process_authentication(PCP_CONNECTION *frontend, char *buf, char* salt, int *random_salt);
+static void send_md5salt(PCP_CONNECTION *frontend, char* salt);
+
+static void pcp_process_command(char tos, char *buf, int buf_len);
+
+static int pool_detach_node(int node_id, bool gracefully);
+static int pool_promote_node(int node_id, bool gracefully);
+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(PCP_CONNECTION *frontend,char mode);
+static void process_set_configration_parameter(PCP_CONNECTION *frontend,char *buf, int len);
+
+static void pcp_worker_will_go_down(int code, Datum arg);
+
+static void do_pcp_flush(PCP_CONNECTION *frontend);
+static void do_pcp_read(PCP_CONNECTION *pc, void *buf, int len);
+
+
+/*
+ * main entry pont of pcp worker child process
+ */
+void
+pcp_worker_main(int port)
+{
+ sigjmp_buf local_sigjmp_buf;
+ MemoryContext PCPMemoryContext;
+ int authenticated = 0;
+
+ char salt[4];
+ int random_salt = 0;
+ struct timeval uptime;
+ char tos;
+ int rsize;
+ char *buf = NULL;
+
+ ereport(DEBUG1,
+ (errmsg("I am PCP worker child with pid:%d",getpid())));
+
+ /* Identify myself via ps */
+ init_ps_display("", "", "", "");
+
+ gettimeofday(&uptime, NULL);
+ srandom((unsigned int) (getpid() ^ uptime.tv_usec));
+
+ /* set up signal handlers */
+ signal(SIGTERM, die);
+ signal(SIGINT, die);
+ signal(SIGQUIT, die);
+ signal(SIGCHLD, SIG_DFL);
+ signal(SIGUSR2, wakeup_handler_child);
+ signal(SIGUSR1, SIG_IGN);
+ signal(SIGHUP, SIG_IGN);
+ signal(SIGPIPE, SIG_IGN);
+ signal(SIGALRM, SIG_IGN);
+ /* Create per loop iteration memory context */
+ PCPMemoryContext = AllocSetContextCreate(TopMemoryContext,
+ "PCP_worker_main_loop",
+ ALLOCSET_DEFAULT_MINSIZE,
+ ALLOCSET_DEFAULT_INITSIZE,
+ ALLOCSET_DEFAULT_MAXSIZE);
+
+ MemoryContextSwitchTo(TopMemoryContext);
+
+ /*
+ * install the call back for preparation of pcp worker child exit
+ */
+ on_system_exit(pcp_worker_will_go_down, (Datum)NULL);
+
+ /* Initialize my backend status */
+ pool_initialize_private_backend_status();
+
+ /* Initialize process context */
+ pool_init_process_context();
+
+ pcp_frontend = pcp_open(port);
+ unset_nonblock(pcp_frontend->fd);
+
+ if (sigsetjmp(local_sigjmp_buf, 1) != 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;
+
+ /* read a PCP packet */
+ do_pcp_read(pcp_frontend, &tos, 1);
+ do_pcp_read(pcp_frontend, &rsize, sizeof(int));
+
+ rsize = ntohl(rsize);
+ if ((rsize - sizeof(int)) > 0)
+ {
+ buf = (char *)palloc(rsize - sizeof(int));
+ do_pcp_read(pcp_frontend, buf, rsize - sizeof(int));
+ }
+
+ ereport(DEBUG1,
+ (errmsg("received PCP packet"),
+ errdetail("PCP packet type of service '%c'", tos)));
+
+ if (tos == 'R') /* authentication */
+ {
+ set_ps_display("PCP: processing authentication", false);
+ process_authentication(pcp_frontend, buf,salt, &random_salt);
+ authenticated = 1;
+ continue;
+ }
+ if (tos == 'M') /* md5 salt */
+ {
+ set_ps_display("PCP: processing authentication", false);
+ send_md5salt(pcp_frontend, salt);
+ random_salt = 1;
+ continue;
+ }
+ /* is this connection authenticated? if not disconnect immediately*/
+ if (!authenticated)
+ ereport(FATAL,
+ (errmsg("authentication failed for new PCP connection"),
+ errdetail("connection not authorized")));
+
+ /* process a request */
+ pcp_process_command(tos, buf, rsize);
+ }
+ exit(0);
+}
+
+
+/* pcp command processor */
+static void
+pcp_process_command(char tos, char *buf, int buf_len)
+{
+
+ if (tos == 'C' || tos == 'd' || tos == 'D' || tos == 'j' ||
+ tos == 'J' || tos == 'O' || tos == 'T')
+ {
+ if (Req_info->switching)
+ {
+ if(Req_info->request_queue_tail != Req_info->request_queue_head)
+ {
+ POOL_REQUEST_KIND reqkind;
+ reqkind = Req_info->request[(Req_info->request_queue_head +1) % MAX_REQUEST_QUEUE_SIZE].kind;
+
+ if (reqkind == NODE_UP_REQUEST)
+ ereport(ERROR,
+ (errmsg("failed to process PCP request at the moment"),
+ errdetail("failback is in progress")));
+ else if (reqkind == NODE_DOWN_REQUEST)
+ ereport(ERROR,
+ (errmsg("failed to process PCP request at the moment"),
+ errdetail("failover is in progress")));
+ else if (reqkind == PROMOTE_NODE_REQUEST)
+ ereport(ERROR,
+ (errmsg("failed to process PCP request at the moment"),
+ errdetail("promote node operation is in progress")));
+
+ ereport(ERROR,
+ (errmsg("failed to process PCP request at the moment"),
+ errdetail("operation is in progress")));
+ }
+ }
+ }
+
+ switch (tos)
+ {
+ case 'A': /* set configuration parameter */
+ set_ps_display("PCP: processing set configration parameter request", false);
+ process_set_configration_parameter(pcp_frontend,buf,buf_len);
+ break;
+
+ case 'L': /* node count */
+ set_ps_display("PCP: processing node count request", false);
+ inform_node_count(pcp_frontend);
+ break;
+
+ case 'I': /* node info */
+ set_ps_display("PCP: processing node info request", false);
+ inform_node_info(pcp_frontend, buf);
+ break;
+
+ case 'N': /* process count */
+ set_ps_display("PCP: processing process count request", false);
+ inform_process_count(pcp_frontend);
+ break;
+
+ case 'P': /* process info */
+ set_ps_display("PCP: processing process info request", false);
+ inform_process_info(pcp_frontend, buf);
+ break;
+
+ case 'S': /* SystemDB info */
+ set_ps_display("PCP: processing systemDB info request", false);
+ inform_systemDB_info(pcp_frontend);
+ break;
+
+ case 'W': /* watchdog info */
+ set_ps_display("PCP: processing watchdog info request", false);
+ inform_watchdog_info(pcp_frontend, buf);
+ break;
+
+ case 'D': /* detach node */
+ case 'd': /* detach node gracefully */
+ set_ps_display("PCP: processingdetach node request", false);
+ process_detach_node(pcp_frontend, buf, tos);
+ break;
+
+ case 'C': /* attach node */
+ set_ps_display("PCP: processing attach node request", false);
+ process_attach_node(pcp_frontend, buf);
+ break;
+
+ case 'T':
+ set_ps_display("PCP: processing shutdown request", false);
+ process_shutown_request(pcp_frontend, buf[0]);
+ break;
+
+ case 'O': /* recovery request */
+ set_ps_display("PCP: processing recovery request", false);
+ process_recovery_request(pcp_frontend, buf);
+ break;
+
+ case 'B': /* status request*/
+ set_ps_display("PCP: processing status request request", false);
+ process_status_request(pcp_frontend);
+ break;
+
+ case 'J': /* promote node */
+ case 'j': /* promote node gracefully */
+ set_ps_display("PCP: processing promote node request", false);
+ process_promote_node(pcp_frontend,buf,tos);
+ break;
+
+ case 'F':
+ ereport(DEBUG1,
+ (errmsg("PCP processing request, stop online recovery")));
+ break;
+
+ case 'X': /* disconnect */
+ ereport(DEBUG1,
+ (errmsg("PCP processing request, client disconnecting"),
+ errdetail("closing PCP connection, and exiting child")));
+ pcp_close(pcp_frontend);
+ pcp_frontend = NULL;
+ /* This child has done its part. Rest in peace now */
+ exit(0);
+ break;
+
+ default:
+ ereport(FATAL,
+ (errmsg("PCP processing request"),
+ errdetail("unknown PCP packet type \"%c\"",tos)));
+ }
+}
+
+static RETSIGTYPE
+die(int sig)
+{
+ ereport(DEBUG1,
+ (errmsg("PCP worker child receives shutdown request signal %d", sig)));
+ if(sig == SIGTERM)
+ {
+ ereport(DEBUG1,
+ (errmsg("PCP worker child receives smart shutdown request."),
+ errdetail("waiting for the child to die its natural death")));
+ }
+ else if (sig == SIGINT)
+ {
+ ereport(DEBUG1,
+ (errmsg("PCP worker child receives fast shutdown request.")));
+ exit(0);
+ }
+ else if (sig == SIGQUIT)
+ {
+ ereport(DEBUG1,
+ (errmsg("PCP worker child receives immediate shutdown request.")));
+ exit(0);
+ }
+ else
+ exit(1);
+}
+
+
+static RETSIGTYPE
+wakeup_handler_child(int sig)
+{
+ pcp_worker_wakeup_request = 1;
+}
+
+/*
+ * unset non-block flag
+ */
+static void
+unset_nonblock(int fd)
+{
+ int var;
+
+ /* set fd to non-blocking */
+ var = fcntl(fd, F_GETFL, 0);
+ if (var == -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)
+ {
+ ereport(FATAL,
+ (errmsg("unable to connect"),
+ errdetail("fcntl system call failed with error : \"%s\"",strerror(errno))));
+ }
+}
+
+/*
+ * see if received username and password matches with one in the file
+ */
+static int
+user_authenticate(char *buf, char *passwd_file, char *salt, int salt_len)
+{
+ FILE *fp = NULL;
+ char packet_username[MAX_USER_PASSWD_LEN+1];
+ char packet_password[MAX_USER_PASSWD_LEN+1];
+ char encrypt_buf[(MD5_PASSWD_LEN+1)*2];
+ char file_username[MAX_USER_PASSWD_LEN+1];
+ char file_password[MAX_USER_PASSWD_LEN+1];
+ char *index = NULL;
+ static char line[MAX_FILE_LINE_LEN+1];
+ int i, len;
+
+ /* strcpy() should be OK, but use strncpy() to be extra careful */
+ strncpy(packet_username, buf, MAX_USER_PASSWD_LEN);
+ index = (char *) memchr(buf, '\0', MAX_USER_PASSWD_LEN);
+ if (index == NULL)
+ {
+ ereport(ERROR,
+ (errmsg("failed to authenticate PCP user"),
+ errdetail("error while reading authentication packet")));
+ return 0;
+ }
+ strncpy(packet_password, ++index, MAX_USER_PASSWD_LEN);
+
+ fp = fopen(passwd_file, "r");
+ if (fp == NULL)
+ {
+ ereport(ERROR,
+ (errmsg("failed to authenticate PCP user"),
+ errdetail("could not open %s. reason: %s", passwd_file, strerror(errno))));
+ return 0;
+ }
+
+ /* for now, I don't care if duplicate username exists in the config file */
+ while ((fgets(line, MAX_FILE_LINE_LEN, fp)) != NULL)
+ {
+ i = 0;
+ len = 0;
+
+ if (line[0] == '\n' || line[0] == '#')
+ continue;
+
+ while (line[i] != ':')
+ {
+ len++;
+ if (++i > MAX_USER_PASSWD_LEN)
+ {
+ fclose(fp);
+ ereport(ERROR,
+ (errmsg("failed to authenticate PCP user"),
+ errdetail("username read from file \"%s\" is larger than maximum allowed username length [%d]", passwd_file, MAX_USER_PASSWD_LEN)));
+ return 0;
+ }
+ }
+ memcpy(file_username, line, len);
+ file_username[len] = '\0';
+
+ if (strcmp(packet_username, file_username) != 0)
+ continue;
+
+ i++;
+ len = 0;
+ while (line[i] != '\n' && line[i] != '\0')
+ {
+ len++;
+ if (++i > MAX_USER_PASSWD_LEN)
+ {
+ fclose(fp);
+ ereport(ERROR,
+ (errmsg("failed to authenticate PCP user"),
+ errdetail("password read from file \"%s\" is larger than maximum allowed password length [%d]", passwd_file, MAX_USER_PASSWD_LEN)));
+ return 0;
+ }
+ }
+
+ memcpy(file_password, line+strlen(file_username)+1, len);
+ file_password[len] = '\0';
+
+ pool_md5_encrypt(file_password, file_username, strlen(file_username),
+ encrypt_buf + MD5_PASSWD_LEN + 1);
+ encrypt_buf[(MD5_PASSWD_LEN+1)*2-1] = '\0';
+
+ pool_md5_encrypt(encrypt_buf+MD5_PASSWD_LEN+1, salt, salt_len,
+ encrypt_buf);
+ encrypt_buf[MD5_PASSWD_LEN] = '\0';
+
+ if (strcmp(encrypt_buf, packet_password) == 0)
+ {
+ fclose(fp);
+ return 1;
+ }
+ }
+ fclose(fp);
+ return 0;
+}
+
+
+/* Dedatch a node */
+static int pool_detach_node(int node_id, bool gracefully)
+{
+ if (!gracefully)
+ {
+ notice_backend_error(node_id); /* send failover request */
+ return 0;
+ }
+
+ /*
+ * Wait until all frontends exit
+ */
+ *InRecovery = RECOVERY_DETACH; /* This wiil ensure that new incoming
+ * connection requests are blocked */
+
+ if (wait_connection_closed())
+ {
+ /* wait timed out */
+ finish_recovery();
+ return -1;
+ }
+
+ /*
+ * Now all frontends have gone. Let's do failover.
+ */
+ notice_backend_error(node_id); /* send failover request */
+
+ /*
+ * Wait for failover completed.
+ */
+ pcp_worker_wakeup_request = 0;
+
+ while (!pcp_worker_wakeup_request)
+ {
+ struct timeval t = {1, 0};
+ select(0, NULL, NULL, NULL, &t);
+ }
+ pcp_worker_wakeup_request = 0;
+
+ /*
+ * Start to accept incoming connections and send SIGUSR2 to pgpool
+ * parent to distribute SIGUSR2 all pgpool children.
+ */
+ finish_recovery();
+
+ return 0;
+}
+
+/* Promote a node */
+static int pool_promote_node(int node_id, bool gracefully)
+{
+ if (!gracefully)
+ {
+ promote_backend(node_id); /* send promote request */
+ return 0;
+ }
+
+ /*
+ * Wait until all frontends exit
+ */
+ *InRecovery = RECOVERY_PROMOTE; /* This wiil ensure that new incoming
+ * connection requests are blocked */
+
+ if (wait_connection_closed())
+ {
+ /* wait timed out */
+ finish_recovery();
+ return -1;
+ }
+
+ /*
+ * Now all frontends have gone. Let's do failover.
+ */
+ promote_backend(node_id); /* send promote request */
+
+ /*
+ * Wait for failover completed.
+ */
+ pcp_worker_wakeup_request = 0;
+
+ while (!pcp_worker_wakeup_request)
+ {
+ struct timeval t = {1, 0};
+ select(0, NULL, NULL, NULL, &t);
+ }
+ pcp_worker_wakeup_request = 0;
+
+ /*
+ * Start to accept incoming connections and send SIGUSR2 to pgpool
+ * parent to distribute SIGUSR2 all pgpool children.
+ */
+ finish_recovery();
+ 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);
+
+ ereport(DEBUG1,
+ (errmsg("PCP: informing process count"),
+ errdetail("%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))
+ {
+ ereport(ERROR,
+ (errmsg("informing process info failed"),
+ errdetail("invalid process ID : %s",buf)));
+ }
+ 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);
+ ereport(DEBUG1,
+ (errmsg("PCP informing process info"),
+ errdetail("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)
+ {
+ ereport(ERROR,
+ (errmsg("informing system db info failed"),
+ errdetail("system DB is not defined")));
+ }
+ else
+ {
+ /* first, send systemDB information and finally the command complete */
+ char port[6];
+ char status[2];
+ char dist_def_num[16];
+ char fin_code[] = "CommandComplete";
+ char code[] = "SystemDBInfo";
+
+ /* 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);
+
+ ereport(DEBUG1,
+ (errmsg("PCP: informing systemDB info"),
+ errdetail("retrieved SystemDB information from shared memory")));
+ }
+}
+
+static void
+inform_watchdog_info(PCP_CONNECTION *frontend,char *buf)
+{
+ int wd_index;
+ int wsize;
+ char code[] = "CommandComplete";
+ char pgpool_port_str[6];
+ char wd_port_str[6];
+ char status[2];
+
+ WdInfo *wi = NULL;
+
+ if (!pool_config->use_watchdog)
+ ereport(ERROR,
+ (errmsg("PCP: informing watchdog info failed"),
+ errdetail("watcdhog is not enabled")));
+
+ wd_index = atoi(buf);
+ wi = wd_get_watchdog_info(wd_index);
+
+ if (wi == NULL)
+ ereport(ERROR,
+ (errmsg("PCP: informing watchdog info failed"),
+ errdetail("invalid watchdog index")));
+
+ ereport(DEBUG2,
+ (errmsg("PCP: informing watchdog info"),
+ errdetail("retrieved node information from shared memory")));
+
+ 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);
+}
+
+static void
+inform_node_info(PCP_CONNECTION *frontend,char *buf)
+{
+ int node_id;
+ int wsize;
+ char port_str[6];
+ char status[2];
+ char weight_str[20];
+ char code[] = "CommandComplete";
+ BackendInfo *bi = NULL;
+
+ node_id = atoi(buf);
+
+ bi = pool_get_node_info(node_id);
+
+ if (bi == NULL)
+ ereport(ERROR,
+ (errmsg("informing node info failed"),
+ errdetail("invalid node ID")));
+
+ ereport(DEBUG2,
+ (errmsg("PCP: informing node info"),
+ errdetail("retrieved node information from shared memory")));
+
+ 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);
+}
+
+static void
+inform_node_count(PCP_CONNECTION *frontend)
+{
+ int wsize;
+ char mesg[16];
+ char code[] = "CommandComplete";
+ int node_count = pool_get_node_count();
+
+ 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);
+
+ ereport(DEBUG1,
+ (errmsg("PCP: informing node count"),
+ errdetail("%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);
+ ereport(DEBUG1,
+ (errmsg("PCP: processing detach node"),
+ errdetail("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);
+ ereport(DEBUG1,
+ (errmsg("PCP: processing attach node"),
+ errdetail("attaching Node ID %d", node_id)));
+
+ send_failback_request(node_id,true);
+
+ 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 wsize;
+ char code[] = "CommandComplete";
+ int node_id = atoi(buf);
+
+ if ( (node_id < 0) || (node_id >= pool_config->backend_desc->num_backends) )
+ ereport(ERROR,
+ (errmsg("process recovery request failed"),
+ 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))
+ {
+ if (MASTER_SLAVE && !strcmp(pool_config->master_slave_sub_mode, MODE_STREAMREP))
+ ereport(ERROR,
+ (errmsg("process recovery request failed"),
+ errdetail("primary server cannot be recovered by online recovery.")));
+ else
+ ereport(ERROR,
+ (errmsg("process recovery request failed"),
+ errdetail("recovery request is accepted only in replication mode or stereaming replication mode.")));
+ }
+ else
+ {
+ if (pcp_mark_recovery_in_progress() == false)
+ ereport(FATAL,
+ (errmsg("process recovery request failed"),
+ errdetail("pgpool-II is already processing another recovery request.")));
+
+ ereport(DEBUG1,
+ (errmsg("PCP: processing recovery request"),
+ errdetail("start online recovery")));
+
+ PG_TRY();
+ {
+ start_recovery(node_id);
+ finish_recovery();
+ 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);
+ pcp_mark_recovery_finished();
+ }
+ PG_CATCH();
+ {
+ finish_recovery();
+ pcp_mark_recovery_finished();
+ PG_RE_THROW();
+
+ }PG_END_TRY();
+ }
+ 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);
+ ereport(DEBUG1,
+ (errmsg("PCP: processing status request"),
+ errdetail("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) )
+ ereport(ERROR,
+ (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))
+ {
+ ereport(FATAL,
+ (errmsg("invalid pgpool mode for process recovery request"),
+ errdetail("not in streaming replication mode, can't promote node id %d", node_id)));
+
+ }
+
+ if (node_id == PRIMARY_NODE_ID)
+ {
+ ereport(FATAL,
+ (errmsg("invalid pgpool mode for process recovery request"),
+ errdetail("specified node is already primary node, can't promote node id %d", node_id)));
+
+ }
+ ereport(DEBUG1,
+ (errmsg("PCP: processing promote node"),
+ errdetail("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* salt, int *random_salt)
+{
+ int wsize;
+ int authenticated;
+
+ if (*random_salt)
+ {
+ authenticated = user_authenticate(buf, pcp_conf_file, salt, 4);
+ }
+ if (!*random_salt || !authenticated)
+ {
+ ereport(FATAL,
+ (errmsg("authentication failed"),
+ errdetail("username and/or password does not match")));
+
+ *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));
+ do_pcp_flush(frontend);
+ *random_salt = 0;
+
+ ereport(DEBUG1,
+ (errmsg("PCP: processing authentication request"),
+ errdetail("authentication OK")));
+ }
+}
+
+static void
+send_md5salt(PCP_CONNECTION *frontend, char* salt)
+{
+ int wsize;
+ ereport(DEBUG1,
+ (errmsg("PCP: sending md5 salt to client")));
+
+ pool_random_salt(salt);
+
+ 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);
+}
+
+static void
+process_shutown_request(PCP_CONNECTION *frontend, char mode)
+{
+ char code[] = "CommandComplete";
+ pid_t ppid = getppid();
+ int sig,len;
+
+ if (mode == 's')
+ {
+ ereport(DEBUG1,
+ (errmsg("PCP: processing shutdown request"),
+ errdetail("sending SIGTERM to the parent process with PID:%d", ppid)));
+ sig = SIGTERM;
+ }
+ else if (mode == 'f')
+ {
+ ereport(DEBUG1,
+ (errmsg("PCP: processing shutdown request"),
+ errdetail("sending SIGINT to the parent process with PID:%d", ppid)));
+ sig = SIGINT;
+ }
+ else if (mode == 'i')
+ {
+ ereport(DEBUG1,
+ (errmsg("PCP: processing shutdown request"),
+ errdetail("sending SIGQUIT to the parent process with PID:%d", ppid)));
+ sig = SIGQUIT;
+ }
+ else
+ {
+ ereport(ERROR,
+ (errmsg("PCP: error while processing shutdown request"),
+ errdetail("invalid shutdown mode \"%c\"", mode)));
+ }
+
+ pcp_write(frontend, "t", 1);
+ len = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &len, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+
+ pool_signal_parent(sig);
+}
+
+static void
+process_set_configration_parameter(PCP_CONNECTION *frontend,char *buf, int len)
+{
+ char* param_name;
+ char* param_value;
+ int wsize;
+ char code[] = "CommandComplete";
+
+ param_name = buf;
+ if(param_name == NULL)
+ ereport(ERROR,
+ (errmsg("PCP: set configuration parameter failed"),
+ errdetail("invalid pcp packet received from client")));
+
+ param_value = (char *) memchr(buf, '\0', len) + 1;
+ if(param_value == NULL)
+ ereport(ERROR,
+ (errmsg("set configuration parameter failed"),
+ errdetail("invalid pcp packet received from client")));
+
+ ereport(LOG,
+ (errmsg("set configuration parameter, \"%s TO %s\"",param_name,param_value)));
+
+ if(strcasecmp(param_name, "client_min_messages") == 0)
+ {
+ const char *ordered_valid_values[] = {"debug5","debug4","debug3","debug2","debug1","log","commerror","info","notice","warning","error",NULL};
+ bool found = false;
+ int i;
+ for(i=0; ; i++)
+ {
+ char* valid_val = (char*)ordered_valid_values[i];
+ if(!valid_val)
+ break;
+
+ if (!strcasecmp(param_value, valid_val))
+ {
+ found = true;
+ pool_config->client_min_messages = i + 10;
+ ereport(DEBUG1,
+ (errmsg("PCP setting parameter \"%s\" to \"%s\"",param_name,param_value)));
+ break;
+ }
+ }
+ if (!found)
+ ereport(ERROR,
+ (errmsg("PCP: set configuration parameter failed"),
+ errdetail("invalid value \"%s\" for parameter \"%s\"",param_value,param_name)));
+ }
+ else
+ ereport(ERROR,
+ (errmsg("PCP: set configuration parameter failed"),
+ errdetail("invalid parameter \"%s\"",param_name)));
+
+ pcp_write(frontend, "a", 1);
+ wsize = htonl(sizeof(code) + sizeof(int));
+ pcp_write(frontend, &wsize, sizeof(int));
+ pcp_write(frontend, code, sizeof(code));
+ do_pcp_flush(frontend);
+}
+/*
+ * 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))));
+}
+
+/*
+ * Wrapper around pcp_read which throws FATAL error when read fails
+ */
+static void
+do_pcp_read(PCP_CONNECTION *pc, void *buf, int len)
+{
+ if (pcp_read(pc,buf,len))
+ ereport(FATAL,
+ (errmsg("unable to read from client"),
+ errdetail("pcp_read failed with error : \"%s\"",strerror(errno))));
+}
+
+int send_to_pcp_frontend(char* data, int len, bool flush)
+{
+ int ret;
+ if (processType != PT_PCP_WORKER || pcp_frontend == NULL)
+ return -1;
+ ret = pcp_write(pcp_frontend, data, len);
+ if (flush && !ret)
+ ret = pcp_flush(pcp_frontend);
+ return ret;
+}
+
+int pcp_frontend_exists(void)
+{
+ if (processType != PT_PCP_WORKER || pcp_frontend == NULL)
+ return -1;
+ return 0;
+}
+
+static void pcp_worker_will_go_down(int code, Datum arg)
+{
+ if (processType != PT_PCP_WORKER)
+ {
+ /* should never happen */
+ ereport(WARNING,
+ (errmsg("pcp_worker_will_go_down called from invalid process")));
+ return;
+ }
+ if(pcp_frontend)
+ pcp_close(pcp_frontend);
+ processState = EXITING;
+ POOL_SETMASK(&UnBlockSig);
+
+}
+
diff --git a/src/pcp_con/recovery.c b/src/pcp_con/recovery.c
index 6b544ac53..e2f17a91d 100644
--- a/src/pcp_con/recovery.c
+++ b/src/pcp_con/recovery.c
@@ -49,7 +49,7 @@ static void check_postmaster_started(BackendInfo *backend);
static char recovery_command[1024];
-extern volatile sig_atomic_t pcp_wakeup_request;
+extern volatile sig_atomic_t pcp_worker_wakeup_request;
/*
* Start online recovery.
@@ -145,14 +145,14 @@ void start_recovery(int recovery_node)
* reset failover completion flag. this is necessary since
* previous failover/failback will set the flag to 1.
*/
- pcp_wakeup_request = 0;
+ pcp_worker_wakeup_request = 0;
/* send failback request to pgpool parent */
send_failback_request(recovery_node,false);
/* wait for failback */
failback_wait_count = 0;
- while (!pcp_wakeup_request)
+ while (!pcp_worker_wakeup_request)
{
struct timeval t = {1, 0};
/* polling SIGUSR2 signal every 1 sec */
@@ -168,7 +168,7 @@ void start_recovery(int recovery_node)
break;
}
}
- pcp_wakeup_request = 0;
+ pcp_worker_wakeup_request = 0;
}
PG_CATCH();
{
diff --git a/src/protocol/child.c b/src/protocol/child.c
index 6f6a13c64..672c9e19a 100644
--- a/src/protocol/child.c
+++ b/src/protocol/child.c
@@ -1033,11 +1033,7 @@ static void send_frontend_exits(void)
int i;
POOL_CONNECTION_POOL *p = pool_connection_pool;
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
POOL_SETMASK2(&BlockSig, &oldmask);
@@ -1554,11 +1550,7 @@ static void s_do_auth(POOL_CONNECTION_POOL_SLOT *cp, char *password)
*/
static void connection_count_up(void)
{
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
POOL_SETMASK2(&BlockSig, &oldmask);
pool_semaphore_lock(CONN_COUNTER_SEM);
@@ -1573,11 +1565,7 @@ static void connection_count_up(void)
*/
static void connection_count_down(void)
{
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
POOL_SETMASK2(&BlockSig, &oldmask);
pool_semaphore_lock(CONN_COUNTER_SEM);
diff --git a/src/protocol/pool_connection_pool.c b/src/protocol/pool_connection_pool.c
index 1f71b8487..047c71162 100644
--- a/src/protocol/pool_connection_pool.c
+++ b/src/protocol/pool_connection_pool.c
@@ -84,11 +84,7 @@ int pool_init_cp(void)
*/
POOL_CONNECTION_POOL *pool_get_cp(char *user, char *database, int protoMajor, int check_socket)
{
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
int i, freed = 0;
ConnectionInfo *info;
diff --git a/src/protocol/pool_process_query.c b/src/protocol/pool_process_query.c
index 7ec8354e9..4ed9f8867 100644
--- a/src/protocol/pool_process_query.c
+++ b/src/protocol/pool_process_query.c
@@ -4103,11 +4103,7 @@ POOL_STATUS end_internal_transaction(POOL_CONNECTION *frontend, POOL_CONNECTION_
int i;
int len;
char tstate;
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
/*
* We must block all signals. If pgpool SIGTERM, SIGINT or SIGQUIT
diff --git a/src/query_cache/pool_memqcache.c b/src/query_cache/pool_memqcache.c
index 9e6849c19..df74df344 100644
--- a/src/query_cache/pool_memqcache.c
+++ b/src/query_cache/pool_memqcache.c
@@ -592,11 +592,7 @@ POOL_STATUS pool_fetch_from_memory_cache(POOL_CONNECTION *frontend,
char *qcache;
size_t qcachelen;
int sts;
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
*foundp = false;
@@ -1780,11 +1776,7 @@ void
pool_clear_memory_cache(void)
{
size_t size;
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
POOL_SETMASK2(&BlockSig, &oldmask);
pool_shmem_lock();
@@ -3096,16 +3088,12 @@ static void pool_check_and_discard_cache_buffer(int num_oids, int *oids)
void pool_handle_query_cache(POOL_CONNECTION_POOL *backend, char *query, Node *node, char state)
{
POOL_SESSION_CONTEXT *session_context;
+ pool_sigset_t oldmask;
char *cache_buffer;
size_t len;
int num_oids;
int *oids;
int i;
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
session_context = pool_get_session_context(true);
@@ -3358,11 +3346,7 @@ int pool_init_memqcache_stats(void)
POOL_QUERY_CACHE_STATS *pool_get_memqcache_stats(void)
{
static POOL_QUERY_CACHE_STATS mystats;
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
memset(&mystats, 0, sizeof(POOL_QUERY_CACHE_STATS));
@@ -3394,11 +3378,7 @@ void pool_reset_memqcache_stats(void)
*/
long long int pool_stats_count_up_num_selects(long long int num)
{
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
POOL_SETMASK2(&BlockSig, &oldmask);
pool_semaphore_lock(QUERY_CACHE_STATS_SEM);
@@ -3449,11 +3429,7 @@ void pool_tmp_stats_reset_num_selects(void)
*/
long long int pool_stats_count_up_num_cache_hits(void)
{
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
POOL_SETMASK2(&BlockSig, &oldmask);
pool_semaphore_lock(QUERY_CACHE_STATS_SEM);
diff --git a/src/test/pgpool_setup b/src/test/pgpool_setup
index 3dae411da..db58f7d96 100755
--- a/src/test/pgpool_setup
+++ b/src/test/pgpool_setup
@@ -199,7 +199,7 @@ then
$pg_ctl -w -m f -D $failed_db_cluster stop >> $log 2>&1
sleep 10
# recovery the node
- pcp_recovery_node 1 localhost $PCP_PORT $pcp_super_user $pcp_super_user $failed_node_id >> $log 2>&1
+ pcp_recovery_node localhost $PCP_PORT $pcp_super_user $pcp_super_user $failed_node_id >> $log 2>&1
else
echo "$failed_db_cluster is not running. skipping follow master command." >> $log
fi
@@ -778,7 +778,7 @@ $PSQL -p $PGPOOL_PORT -c "show pool_nodes" test
while [ $n -lt $NUMCLUSTERS ]
do
echo -n "recovery node $n..."
- $PGPOOL_INSTALL_DIR/bin/pcp_recovery_node 1 localhost $PCP_PORT $WHOAMI $WHOAMI $n
+ $PGPOOL_INSTALL_DIR/bin/pcp_recovery_node localhost $PCP_PORT $WHOAMI $WHOAMI $n
echo "done."
n=`expr $n + 1`
if [ $n -lt $NUMCLUSTERS ];then
diff --git a/src/test/regression/tests/054.postgres_fdw/test.sh b/src/test/regression/tests/054.postgres_fdw/test.sh
index 0eba9d7fd..dfa977d77 100755
--- a/src/test/regression/tests/054.postgres_fdw/test.sh
+++ b/src/test/regression/tests/054.postgres_fdw/test.sh
@@ -43,7 +43,7 @@ EOF
# detach node #1 so that pgpool does not access
# postgres_fdw. Otherwise it will go into an inifinite loop.
-$PGPOOL_INSTALL_DIR/bin/pcp_detach_node 1 localhost $PCP_PORT $WHOAMI $WHOAMI 1
+$PGPOOL_INSTALL_DIR/bin/pcp_detach_node localhost $PCP_PORT $WHOAMI $WHOAMI 1
if [ $? != 0 ];then
echo "pcp_detach_node failed"
diff --git a/src/tools/pcp/pcp_attach_node.c b/src/tools/pcp/pcp_attach_node.c
index 6ea273a62..55686aec8 100644
--- a/src/tools/pcp/pcp_attach_node.c
+++ b/src/tools/pcp/pcp_attach_node.c
@@ -38,7 +38,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -71,35 +70,29 @@ main(int argc, char **argv)
argc -= optind;
argv += optind;
- if (argc != 6)
+ if (argc != 5)
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(user, argv[3]);
+ strcpy(user, argv[2]);
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(pass, argv[4]);
+ strcpy(pass, argv[3]);
- nodeID = atoi(argv[5]);
+ nodeID = atoi(argv[4]);
if (nodeID < 0 || nodeID > MAX_NUM_BACKENDS)
myexit(NULL);
- pcp_set_timeout(timeout);
-
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
myexit(pcpConn);
@@ -118,10 +111,9 @@ static void
usage(void)
{
fprintf(stderr, "pcp_attach_node - attach a node from pgpool-II\n\n");
- fprintf(stderr, "Usage: pcp_attach_node [-d] timeout hostname port# username password nodeID\n");
+ fprintf(stderr, "Usage: pcp_attach_node [-d] hostname port# username password nodeID\n");
fprintf(stderr, "Usage: pcp_attach_node -h\n\n");
fprintf(stderr, " -d, --debug : enable debug message (optional)\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pcp/pcp_detach_node.c b/src/tools/pcp/pcp_detach_node.c
index be6eec000..5aca3d990 100644
--- a/src/tools/pcp/pcp_detach_node.c
+++ b/src/tools/pcp/pcp_detach_node.c
@@ -38,7 +38,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -77,34 +76,28 @@ main(int argc, char **argv)
argc -= optind;
argv += optind;
- if (argc != 6)
+ if (argc != 5)
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(user, argv[3]);
+ strcpy(user, argv[2]);
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(pass, argv[4]);
+ strcpy(pass, argv[3]);
- nodeID = atoi(argv[5]);
+ nodeID = atoi(argv[4]);
if (nodeID < 0 || nodeID > MAX_NUM_BACKENDS)
myexit(NULL);
-
- pcp_set_timeout(timeout);
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
@@ -128,11 +121,10 @@ static void
usage(void)
{
fprintf(stderr, "pcp_detach_node - detach a node from pgpool-II\n\n");
- fprintf(stderr, "Usage: pcp_detach_node [-d][-g] timeout hostname port# username password nodeID\n");
+ fprintf(stderr, "Usage: pcp_detach_node [-d][-g] hostname port# username password nodeID\n");
fprintf(stderr, "Usage: pcp_detach_node -h\n\n");
fprintf(stderr, " -d, --debug : enable debug message (optional)\n");
fprintf(stderr, " -g, --gracefully : detach gracefully(optional)\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pcp/pcp_node_count.c b/src/tools/pcp/pcp_node_count.c
index a71d3059d..514c4c2ff 100644
--- a/src/tools/pcp/pcp_node_count.c
+++ b/src/tools/pcp/pcp_node_count.c
@@ -38,7 +38,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -71,31 +70,24 @@ main(int argc, char **argv)
argc -= optind;
argv += optind;
- if (argc != 5)
+ if (argc != 4)
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
-
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(user, argv[3]);
+ strcpy(user, argv[2]);
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(pass, argv[4]);
-
- pcp_set_timeout(timeout);
+ strcpy(pass, argv[3]);
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
@@ -118,10 +110,9 @@ static void
usage(void)
{
fprintf(stderr, "pcp_node_count - display the total number of nodes under pgpool-II's control\n\n");
- fprintf(stderr, "Usage: pcp_node_count [-d] timeout hostname port# username password\n");
+ fprintf(stderr, "Usage: pcp_node_count [-d] hostname port# username password\n");
fprintf(stderr, "Usage: pcp_node_count -h\n\n");
fprintf(stderr, " -d, --debug : enable debug message (optional)\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pcp/pcp_node_info.c b/src/tools/pcp/pcp_node_info.c
index f317341c8..bddcdce3a 100644
--- a/src/tools/pcp/pcp_node_info.c
+++ b/src/tools/pcp/pcp_node_info.c
@@ -38,7 +38,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -78,34 +77,29 @@ main(int argc, char **argv)
argc -= optind;
argv += optind;
- if (argc != 6)
+ if (argc != 5)
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(user, argv[3]);
+ strcpy(user, argv[2]);
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(pass, argv[4]);
+ strcpy(pass, argv[3]);
- nodeID = atoi(argv[5]);
+ nodeID = atoi(argv[4]);
if (nodeID < 0 || nodeID > MAX_NUM_BACKENDS)
myexit(NULL);
- pcp_set_timeout(timeout);
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
myexit(pcpConn);
@@ -140,9 +134,8 @@ static void
usage(void)
{
fprintf(stderr, "pcp_node_info - display a pgpool-II node's information\n\n");
- fprintf(stderr, "Usage: pcp_node_info [-d] timeout hostname port# username password nodeID\n");
+ fprintf(stderr, "Usage: pcp_node_info [-d] hostname port# username password nodeID\n");
fprintf(stderr, " -d, --debug : enable debug message (optional)\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pcp/pcp_pool_status.c b/src/tools/pcp/pcp_pool_status.c
index 7de970af0..55facdf35 100644
--- a/src/tools/pcp/pcp_pool_status.c
+++ b/src/tools/pcp/pcp_pool_status.c
@@ -38,7 +38,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -74,31 +73,24 @@ main(int argc, char **argv)
argc -= optind;
argv += optind;
- if (argc != 5)
+ if (argc != 4)
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
+ strcpy(user, argv[2]);
- strcpy(user, argv[3]);
-
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(pass, argv[4]);
-
- pcp_set_timeout(timeout);
+ strcpy(pass, argv[3]);
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
@@ -130,10 +122,9 @@ static void
usage(void)
{
fprintf(stderr, "pcp_pool_status - display pgpool configuration and status\n\n");
- fprintf(stderr, "Usage: pcp_pool_status [-d] timeout hostname port# username password\n");
+ fprintf(stderr, "Usage: pcp_pool_status [-d] hostname port# username password\n");
fprintf(stderr, "Usage: pcp_pool_status -h\n\n");
fprintf(stderr, " -d, --debug : enable debug message (optional)\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pcp/pcp_proc_count.c b/src/tools/pcp/pcp_proc_count.c
index c7c8b9164..fa915da93 100644
--- a/src/tools/pcp/pcp_proc_count.c
+++ b/src/tools/pcp/pcp_proc_count.c
@@ -38,7 +38,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -73,30 +72,24 @@ main(int argc, char **argv)
argc -= optind;
argv += optind;
- if (argc != 5)
+ if (argc != 4)
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(user, argv[3]);
+ strcpy(user, argv[2]);
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(pass, argv[4]);
-
- pcp_set_timeout(timeout);
+ strcpy(pass, argv[3]);
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
@@ -124,9 +117,8 @@ static void
usage(void)
{
fprintf(stderr, "pcp_proc_count - display the list of pgpool-II child process PIDs\n\n");
- fprintf(stderr, "Usage: pcp_proc_count timeout hostname port# username password\n");
+ fprintf(stderr, "Usage: pcp_proc_count hostname port# username password\n");
fprintf(stderr, "Usage: pcp_node_info -h\n\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pcp/pcp_proc_info.c b/src/tools/pcp/pcp_proc_info.c
index 9be65930e..2692ce2ce 100644
--- a/src/tools/pcp/pcp_proc_info.c
+++ b/src/tools/pcp/pcp_proc_info.c
@@ -39,7 +39,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -119,32 +118,28 @@ main(int argc, char **argv)
frmt = "%s %s %s %s %d %d %d %d %d\n";
}
- if (!(argc == 5 || argc == 6))
+ if (!(argc == 4 || argc == 5))
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(user, argv[3]);
+ strcpy(user, argv[2]);
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(pass, argv[4]);
+ strcpy(pass, argv[3]);
- if (argc == 6) {
+ if (argc == 5) {
- processID = atoi(argv[5]);
+ processID = atoi(argv[4]);
if (processID < 0)
myexit(NULL);
}
@@ -152,8 +147,6 @@ main(int argc, char **argv)
processID = 0;
}
- pcp_set_timeout(timeout);
-
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
myexit(pcpConn);
@@ -208,9 +201,8 @@ static void
usage(void)
{
fprintf(stderr, "pcp_proc_info - display a pgpool-II child process' information\n\n");
- fprintf(stderr, "Usage: pcp_proc_info [-d] timeout hostname port# username password [PID]\n");
+ fprintf(stderr, "Usage: pcp_proc_info [-d] hostname port# username password [PID]\n");
fprintf(stderr, " -d, --debug : enable debug message (optional)\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pcp/pcp_promote_node.c b/src/tools/pcp/pcp_promote_node.c
index 676e84dd5..1fef4e0a2 100644
--- a/src/tools/pcp/pcp_promote_node.c
+++ b/src/tools/pcp/pcp_promote_node.c
@@ -38,7 +38,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -79,34 +78,28 @@ main(int argc, char **argv)
argc -= optind;
argv += optind;
- if (argc != 6)
+ if (argc != 5)
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(user, argv[3]);
+ strcpy(user, argv[2]);
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(pass, argv[4]);
+ strcpy(pass, argv[3]);
- nodeID = atoi(argv[5]);
+ nodeID = atoi(argv[4]);
if (nodeID < 0 || nodeID > MAX_NUM_BACKENDS)
myexit(NULL);
-
- pcp_set_timeout(timeout);
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
@@ -130,11 +123,10 @@ static void
usage(void)
{
fprintf(stderr, "pcp_promote_node - promote a node as new master from pgpool-II\n\n");
- fprintf(stderr, "Usage: pcp_promote_node [-d][-g] timeout hostname port# username password nodeID\n");
+ fprintf(stderr, "Usage: pcp_promote_node [-d][-g] hostname port# username password nodeID\n");
fprintf(stderr, "Usage: pcp_promote_node -h\n\n");
fprintf(stderr, " -d, --debug : enable debug message (optional)\n");
fprintf(stderr, " -g, --gracefully : promote gracefully(optional)\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pcp/pcp_recovery_node.c b/src/tools/pcp/pcp_recovery_node.c
index dd9cead6e..90c37e56e 100644
--- a/src/tools/pcp/pcp_recovery_node.c
+++ b/src/tools/pcp/pcp_recovery_node.c
@@ -38,7 +38,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -73,35 +72,29 @@ main(int argc, char **argv)
argc -= optind;
argv += optind;
- if (argc != 6)
+ if (argc != 5)
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(user, argv[3]);
+ strcpy(user, argv[2]);
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(pass, argv[4]);
+ strcpy(pass, argv[3]);
- nodeID = atoi(argv[5]);
+ nodeID = atoi(argv[4]);
if (nodeID < 0 || nodeID > MAX_NUM_BACKENDS)
myexit(NULL);
- pcp_set_timeout(timeout);
-
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
myexit(pcpConn);
@@ -120,10 +113,9 @@ static void
usage(void)
{
fprintf(stderr, "pcp_recovery_node - recovery a node\n\n");
- fprintf(stderr, "Usage: pcp_recovery_node [-d] timeout hostname port# username password nodeID\n");
+ fprintf(stderr, "Usage: pcp_recovery_node [-d] hostname port# username password nodeID\n");
fprintf(stderr, "Usage: pcp_recovery_node -h\n\n");
fprintf(stderr, " -d, --debug : enable debug message (optional)\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pcp/pcp_stop_pgpool.c b/src/tools/pcp/pcp_stop_pgpool.c
index 16e66cf38..5d5b8a24d 100644
--- a/src/tools/pcp/pcp_stop_pgpool.c
+++ b/src/tools/pcp/pcp_stop_pgpool.c
@@ -38,7 +38,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -71,43 +70,32 @@ main(int argc, char **argv)
argc -= optind;
argv += optind;
- if (argc != 6)
+ if (argc != 5)
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
-
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
-
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
-
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
+ strcpy(user, argv[2]);
- strcpy(user, argv[3]);
-
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
+ strcpy(pass, argv[3]);
- strcpy(pass, argv[4]);
-
- if (strlen(argv[5]) != 1)
+ if (strlen(argv[4]) != 1)
myexit(NULL);
- mode = argv[5][0];
+ mode = argv[4][0];
if (mode != 's' && mode != 'f' && mode != 'i')
myexit(NULL);
- pcp_set_timeout(timeout);
-
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
myexit(pcpConn);
@@ -126,10 +114,9 @@ static void
usage(void)
{
fprintf(stderr, "pcp_stop_pgpool - terminate pgpool-II\n\n");
- fprintf(stderr, "Usage: pcp_stop_pgpool [-d] timeout hostname port# username password mode\n");
+ fprintf(stderr, "Usage: pcp_stop_pgpool [-d] hostname port# username password mode\n");
fprintf(stderr, "Usage: pcp_stop_pgpool -h\n\n");
fprintf(stderr, " -d, --debug : enable debug message (optional)\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pcp/pcp_systemdb_info.c b/src/tools/pcp/pcp_systemdb_info.c
index 56179645e..91b8a4267 100644
--- a/src/tools/pcp/pcp_systemdb_info.c
+++ b/src/tools/pcp/pcp_systemdb_info.c
@@ -38,7 +38,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -74,28 +73,24 @@ main(int argc, char **argv)
argc -= optind;
argv += optind;
- if (argc != 5)
+ if (argc != 4)
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(user, argv[3]);
+ strcpy(user, argv[2]);
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(pass, argv[4]);
+ strcpy(pass, argv[3]);
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
@@ -146,10 +141,9 @@ static void
usage(void)
{
fprintf(stderr, "pcp_systemdb_info - display the pgpool-II systemDB information\n\n");
- fprintf(stderr, "Usage: pcp_systemdb_info [-d] timeout hostname port# username password\n");
+ fprintf(stderr, "Usage: pcp_systemdb_info [-d] hostname port# username password\n");
fprintf(stderr, "Usage: pcp_systemdb_info -h\n\n");
fprintf(stderr, " -d, --debug : enable debug message (optional)\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pcp/pcp_watchdog_info.c b/src/tools/pcp/pcp_watchdog_info.c
index a4d4623f6..f1ae138bc 100644
--- a/src/tools/pcp/pcp_watchdog_info.c
+++ b/src/tools/pcp/pcp_watchdog_info.c
@@ -38,7 +38,6 @@ static void myexit(PCPConnInfo* pcpConn);
int
main(int argc, char **argv)
{
- long timeout;
char host[MAX_DB_HOST_NAMELEN];
int port;
char user[MAX_USER_PASSWD_LEN];
@@ -80,32 +79,28 @@ main(int argc, char **argv)
argc -= optind;
argv += optind;
- if (!(argc == 5 || argc == 6))
+ if (!(argc == 4 || argc == 5))
myexit(NULL);
- timeout = atol(argv[0]);
- if (timeout < 0)
+ if (strlen(argv[0]) >= MAX_DB_HOST_NAMELEN)
myexit(NULL);
+ strcpy(host, argv[0]);
- if (strlen(argv[1]) >= MAX_DB_HOST_NAMELEN)
- myexit(NULL);
- strcpy(host, argv[1]);
-
- port = atoi(argv[2]);
+ port = atoi(argv[1]);
if (port <= 1024 || port > 65535)
myexit(NULL);
- if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[2]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(user, argv[3]);
+ strcpy(user, argv[2]);
- if (strlen(argv[4]) >= MAX_USER_PASSWD_LEN)
+ if (strlen(argv[3]) >= MAX_USER_PASSWD_LEN)
myexit(NULL);
- strcpy(pass, argv[4]);
+ strcpy(pass, argv[3]);
- if (argc == 6)
+ if (argc == 5)
{
- wd_index = atoi(argv[5]);
+ wd_index = atoi(argv[4]);
if (wd_index < 0 || wd_index > MAX_WATCHDOG_NUM)
myexit(NULL);
wd_index++;
@@ -115,8 +110,6 @@ main(int argc, char **argv)
wd_index = 0;
}
- pcp_set_timeout(timeout);
-
pcpConn = pcp_connect(host, port, user, pass, debug?stdout:NULL);
if(PCPConnectionStatus(pcpConn) != PCP_CONNECTION_OK)
myexit(pcpConn);
@@ -152,9 +145,8 @@ static void
usage(void)
{
fprintf(stderr, "pcp_watchdog_info - display a pgpool-II watchdog's information\n\n");
- fprintf(stderr, "Usage: pcp_watchdog_info [-d] timeout hostname port# username password [watchdogID]\n");
+ fprintf(stderr, "Usage: pcp_watchdog_info [-d] hostname port# username password [watchdogID]\n");
fprintf(stderr, " -d, --debug : enable debug message (optional)\n");
- fprintf(stderr, " timeout : connection timeout value in seconds. command exits on timeout\n");
fprintf(stderr, " hostname : pgpool-II hostname\n");
fprintf(stderr, " port# : PCP port number\n");
fprintf(stderr, " username : username for PCP authentication\n");
diff --git a/src/tools/pgmd5/pool_config.c b/src/tools/pgmd5/pool_config.c
index 5de8ea2b9..0bf9db8fd 100644
--- a/src/tools/pgmd5/pool_config.c
+++ b/src/tools/pgmd5/pool_config.c
@@ -499,7 +499,7 @@ char *yytext;
* pgpool: a language independent connection pool server for PostgreSQL
* written by Tatsuo Ishii
*
- * Copyright (c) 2003-2014 PgPool Global Development Group
+ * Copyright (c) 2003-2015 PgPool Global Development Group
*
* Permission to use, copy, modify, and distribute this software and
* its documentation for any purpose and without fee is hereby
@@ -553,8 +553,9 @@ typedef enum {
static char *extract_string(char *value, POOL_TOKEN token);
static char **extract_string_tokens(char *str, char *delim, int *n);
static void clear_host_entry(int slot);
+static bool check_redirect_node_spec(char *node_spec);
-#line 558 "pool_config.c"
+#line 559 "pool_config.c"
#define INITIAL 0
@@ -739,10 +740,10 @@ YY_DECL
register char *yy_cp, *yy_bp;
register int yy_act;
-#line 89 "pool_config.l"
+#line 90 "pool_config.l"
-#line 746 "pool_config.c"
+#line 747 "pool_config.c"
if ( !(yy_init) )
{
@@ -824,12 +825,12 @@ do_action: /* This label is used only to access EOF actions. */
case 1:
/* rule 1 can match eol */
YY_RULE_SETUP
-#line 91 "pool_config.l"
+#line 92 "pool_config.l"
Lineno++; return POOL_EOL;
YY_BREAK
case 2:
YY_RULE_SETUP
-#line 92 "pool_config.l"
+#line 93 "pool_config.l"
/* eat whitespace */
YY_BREAK
case 3:
@@ -837,50 +838,50 @@ case 3:
(yy_c_buf_p) = yy_cp -= 1;
YY_DO_BEFORE_ACTION; /* set up yytext again */
YY_RULE_SETUP
-#line 93 "pool_config.l"
+#line 94 "pool_config.l"
/* eat comment */
YY_BREAK
case 4:
YY_RULE_SETUP
-#line 95 "pool_config.l"
+#line 96 "pool_config.l"
return POOL_KEY;
YY_BREAK
case 5:
YY_RULE_SETUP
-#line 96 "pool_config.l"
+#line 97 "pool_config.l"
return POOL_STRING;
YY_BREAK
case 6:
YY_RULE_SETUP
-#line 97 "pool_config.l"
+#line 98 "pool_config.l"
return POOL_UNQUOTED_STRING;
YY_BREAK
case 7:
YY_RULE_SETUP
-#line 98 "pool_config.l"
+#line 99 "pool_config.l"
return POOL_INTEGER;
YY_BREAK
case 8:
YY_RULE_SETUP
-#line 99 "pool_config.l"
+#line 100 "pool_config.l"
return POOL_REAL;
YY_BREAK
case 9:
YY_RULE_SETUP
-#line 100 "pool_config.l"
+#line 101 "pool_config.l"
return POOL_EQUALS;
YY_BREAK
case 10:
YY_RULE_SETUP
-#line 102 "pool_config.l"
+#line 103 "pool_config.l"
return POOL_PARSE_ERROR;
YY_BREAK
case 11:
YY_RULE_SETUP
-#line 104 "pool_config.l"
+#line 105 "pool_config.l"
ECHO;
YY_BREAK
-#line 884 "pool_config.c"
+#line 885 "pool_config.c"
case YY_STATE_EOF(INITIAL):
yyterminate();
@@ -1838,7 +1839,7 @@ void yyfree (void * ptr )
#define YYTABLES_NAME "yytables"
-#line 104 "pool_config.l"
+#line 105 "pool_config.l"
@@ -1871,7 +1872,6 @@ int pool_init_config(void)
pool_config->socket_dir = DEFAULT_SOCKET_DIR;
pool_config->pcp_socket_dir = DEFAULT_SOCKET_DIR;
pool_config->backend_socket_dir = NULL;
- pool_config->pcp_timeout = 10;
pool_config->num_init_children = 32;
pool_config->listen_backlog_multiplier = 2;
pool_config->max_pool = 4;
@@ -1904,7 +1904,7 @@ int pool_init_config(void)
pool_config->num_white_function_list = 0;
pool_config->black_function_list = default_black_function_list;
pool_config->num_black_function_list = sizeof(default_black_function_list)/sizeof(char *);
- pool_config->log_line_prefix = "";
+ pool_config->log_line_prefix = "%t: pid %p: ";
pool_config->log_error_verbosity = 1; /* PGERROR_DEFAULT */
pool_config->client_min_messages = 18; /* NOTICE */
pool_config->log_min_messages = 19; /* WARNING */
@@ -2366,21 +2366,6 @@ int pool_get_config(char *confpath, POOL_CONFIG_CONTEXT context)
}
pool_config->pcp_socket_dir = str;
}
- else if (!strcmp(key, "pcp_timeout") &&
- CHECK_CONTEXT(INIT_CONFIG|RELOAD_CONFIG, context))
- {
- int v = atoi(yytext);
-
- if (token != POOL_INTEGER || v < 0)
- {
- fclose(fd);
- ereport(error_level,
- (errmsg("invalid configuration for key \"%s\"",key),
- errdetail("invalid value:\"%s\" for key:\"%s\". %s must be >= 0",yytext,key, key)));
- return(-1);
- }
- pool_config->pcp_timeout = v;
- }
else if (!strcmp(key, "num_init_children") && CHECK_CONTEXT(INIT_CONFIG, context))
{
int v = atoi(yytext);
@@ -5039,7 +5024,17 @@ int pool_get_config(char *confpath, POOL_CONFIG_CONTEXT context)
for (i=0;i<lrtokens->pos;i++)
{
- if (add_regex_array(pool_config->redirect_dbnames, lrtokens->token[i].left_token))
+ if (!check_redirect_node_spec(lrtokens->token[i].right_token))
+ {
+ fclose(fd);
+ ereport(error_level,
+ (errmsg("invalid configuration for key \"%s\"",key),
+ errdetail("wrong redirect db node spec: \"%s\"", lrtokens->token[i].right_token)));
+ return(-1);
+ }
+
+ if (*(lrtokens->token[i].left_token) == '\0' ||
+ add_regex_array(pool_config->redirect_dbnames, lrtokens->token[i].left_token))
{
fclose(fd);
ereport(error_level,
@@ -5079,12 +5074,23 @@ int pool_get_config(char *confpath, POOL_CONFIG_CONTEXT context)
for (i=0;i<lrtokens->pos;i++)
{
- if (add_regex_array(pool_config->redirect_app_names, lrtokens->token[i].left_token))
+ if (!check_redirect_node_spec(lrtokens->token[i].right_token))
{
fclose(fd);
ereport(error_level,
(errmsg("invalid configuration for key \"%s\"",key),
- errdetail("wrong redirect dbname regular expression: \"%s\"", lrtokens->token[i].left_token)));
+ errdetail("wrong redirect db node spec: \"%s\"", lrtokens->token[i].right_token)));
+ return(-1);
+ }
+
+
+ if (*(lrtokens->token[i].left_token) == '\0' ||
+ add_regex_array(pool_config->redirect_app_names, lrtokens->token[i].left_token))
+ {
+ fclose(fd);
+ ereport(error_level,
+ (errmsg("invalid configuration for key \"%s\"",key),
+ errdetail("wrong redirect app name regular expression: \"%s\"", lrtokens->token[i].left_token)));
return(-1);
}
}
@@ -5486,3 +5492,40 @@ char *pool_flag_to_str(unsigned short flag)
return buf;
}
+/*
+ * Check DB node spec. node spec should be either "primary", "standby" or
+ * numeric DB node id.
+*/
+bool check_redirect_node_spec(char *node_spec)
+{
+ int len = strlen(node_spec);
+ int i;
+ long val;
+
+ if (len <= 0)
+ return false;
+
+ if (strcasecmp("primary", node_spec) == 0)
+ {
+ return true;
+ }
+
+ if (strcasecmp("standby", node_spec) == 0)
+ {
+ return true;
+ }
+
+ for (i=0;i<len;i++)
+ {
+ if (!isdigit((int)node_spec[i]))
+ return false;
+ }
+
+ val = atol(node_spec);
+
+ if (val >=0 && val < MAX_NUM_BACKENDS)
+ return true;
+
+ return false;
+}
+
diff --git a/src/utils/error/elog.c b/src/utils/error/elog.c
index 834b7b31d..879863d5d 100644
--- a/src/utils/error/elog.c
+++ b/src/utils/error/elog.c
@@ -2331,6 +2331,9 @@ process_name(void)
case PT_PCP:
prefix = _("PCP CHILD");
break;
+ case PT_PCP_WORKER:
+ prefix = _("PCP WORKER");
+ break;
case PT_HB_SENDER:
prefix = _("WD HB SENDER");
break;
diff --git a/src/utils/pcp/pcp_error.c b/src/utils/pcp/pcp_error.c
deleted file mode 100644
index 552321855..000000000
--- a/src/utils/pcp/pcp_error.c
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * $Header$
- *
- * pgpool: a language independent connection pool server for PostgreSQL
- * written by Tatsuo Ishii
- *
- * Copyright (c) 2003-2008 PgPool Global Development Group
- *
- * Permission to use, copy, modify, and distribute this software and
- * its documentation for any purpose and without fee is hereby
- * granted, provided that the above copyright notice appear in all
- * copies and that both that copyright notice and this permission
- * notice appear in supporting documentation, and that the name of the
- * author not be used in advertising or publicity pertaining to
- * distribution of the software without specific, written prior
- * permission. The author makes no representations about the
- * suitability of this software for any purpose. It is provided "as
- * is" without express or implied warranty.
- *
- * Handles errors occured in PCP modules.
- */
-
-#include <stdio.h>
-
-#include "pcp/pcp.h"
-
-ErrorCode errorcode;
-
-void
-pcp_errorstr(ErrorCode e)
-{
- switch (e)
- {
- case EOFERR:
- fprintf(stdout, "EOFError\n");
- break;
-
- case NOMEMERR:
- fprintf(stdout, "NoMemoryError\n");
- break;
-
- case READERR:
- fprintf(stdout, "ReadError\n");
- break;
-
- case WRITEERR:
- fprintf(stdout, "WriteError\n");
- break;
-
- case TIMEOUTERR:
- fprintf(stdout, "TimeoutError\n");
- break;
-
- case INVALERR:
- fprintf(stdout, "InvalidArgumentError\n");
- break;
-
- case CONNERR:
- fprintf(stdout, "ConnectionError\n");
- break;
-
- case NOCONNERR:
- fprintf(stdout, "NoConnectionError\n");
- break;
-
- case SOCKERR:
- fprintf(stdout, "SocketError\n");
- break;
-
- case HOSTERR:
- fprintf(stdout, "HostError\n");
- break;
-
- case BACKENDERR:
- fprintf(stdout, "BackendError\n");
- break;
-
- case AUTHERR:
- fprintf(stdout, "AuthorizationError\n");
- break;
-
- case UNKNOWNERR:
- default:
- fprintf(stdout, "UnknownError\n");
- break;
- }
-}
diff --git a/src/utils/pcp/pcp_stream.c b/src/utils/pcp/pcp_stream.c
index b0a0f5989..5d95964e3 100644
--- a/src/utils/pcp/pcp_stream.c
+++ b/src/utils/pcp/pcp_stream.c
@@ -41,7 +41,7 @@
static int consume_pending_data(PCP_CONNECTION *pc, void *data, int len);
static int save_pending_data(PCP_CONNECTION *pc, void *data, int len);
-static int pcp_check_fd(PCP_CONNECTION *pc, int notimeout);
+static int pcp_check_fd(PCP_CONNECTION *pc);
/* --------------------------------
* pcp_open - allocate read & write buffers for PCP_CONNECTION
@@ -104,7 +104,6 @@ pcp_read(PCP_CONNECTION *pc, void *buf, int len)
int consume_size;
int readlen;
- int notimeout = 0;
consume_size = consume_pending_data(pc, buf, len);
len -= consume_size;
@@ -112,11 +111,8 @@ pcp_read(PCP_CONNECTION *pc, void *buf, int len)
while (len > 0)
{
- if (pcp_check_fd(pc, notimeout))
- {
- errorcode = TIMEOUTERR;
+ if (pcp_check_fd(pc))
return -1;
- }
readlen = read(pc->fd, readbuf, READBUFSZ);
if (readlen == -1)
@@ -124,12 +120,10 @@ pcp_read(PCP_CONNECTION *pc, void *buf, int len)
if (errno == EAGAIN || errno == EINTR)
continue;
- errorcode = READERR;
return -1;
}
else if (readlen == 0)
{
- errorcode = EOFERR;
return -1;
}
@@ -137,10 +131,7 @@ pcp_read(PCP_CONNECTION *pc, void *buf, int len)
{
/* overrun. we need to save remaining data to pending buffer */
if (save_pending_data(pc, readbuf+len, readlen-len))
- {
- errorcode = NOMEMERR;
return -1;
- }
memmove(buf, readbuf, len);
break;
}
@@ -166,7 +157,6 @@ pcp_write(PCP_CONNECTION *pc, void *buf, int len)
if (len < 0)
{
- errorcode = INVALERR;
return -1;
}
@@ -237,7 +227,6 @@ pcp_flush(PCP_CONNECTION *pc)
else if (wlen < 0)
{
- errorcode = WRITEERR;
return -1;
}
@@ -249,14 +238,9 @@ pcp_flush(PCP_CONNECTION *pc)
}
}
else if (errno == EAGAIN || errno == EINTR)
- {
continue;
- }
else
- {
- errorcode = WRITEERR;
return -1;
- }
}
pc->wbufpo = 0;
@@ -341,14 +325,12 @@ save_pending_data(PCP_CONNECTION *pc, void *data, int len)
* --------------------------------
*/
static int
-pcp_check_fd(PCP_CONNECTION *pc, int notimeout)
+pcp_check_fd(PCP_CONNECTION *pc)
{
fd_set readmask;
fd_set exceptmask;
int fd;
int fds;
- struct timeval timeout;
- struct timeval *tp;
fd = pc->fd;
@@ -359,17 +341,7 @@ pcp_check_fd(PCP_CONNECTION *pc, int notimeout)
FD_SET(fd, &readmask);
FD_SET(fd, &exceptmask);
- if (notimeout || (pcp_timeout.tv_sec + pcp_timeout.tv_usec) == 0)
- tp = NULL;
- else
- {
- /***** haven't got timeout option yet. hard-code it *****/
- timeout.tv_sec = pcp_timeout.tv_sec;
- timeout.tv_usec = pcp_timeout.tv_usec;
- tp = &timeout;
- }
-
- fds = select(fd+1, &readmask, NULL, &exceptmask, tp);
+ fds = select(fd+1, &readmask, NULL, &exceptmask, NULL);
if (fds == -1)
{
diff --git a/src/utils/pcp/pcp_timeout.c b/src/utils/pcp/pcp_timeout.c
deleted file mode 100644
index f4399ecfa..000000000
--- a/src/utils/pcp/pcp_timeout.c
+++ /dev/null
@@ -1,12 +0,0 @@
-#include <sys/time.h>
-
-struct timeval pcp_timeout;
-
-void
-pcp_set_timeout(long sec)
-{
- /* disable timeout (wait forever!) (2008/02/08 yamaguti) */
- sec = 0;
- pcp_timeout.tv_sec = sec;
-}
-
diff --git a/src/utils/pool_process_reporting.c b/src/utils/pool_process_reporting.c
index 49ee11d65..77128bc85 100644
--- a/src/utils/pool_process_reporting.c
+++ b/src/utils/pool_process_reporting.c
@@ -1628,11 +1628,7 @@ void cache_reporting(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend)
static unsigned char nullmap[2] = {0xff, 0xff};
int nbytes = (num_fields + 7)/8;
volatile POOL_SHMEM_STATS *mystats;
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
+ pool_sigset_t oldmask;
double ratio;
#define POOL_CACHE_STATS_MAX_STRING_LEN 32
diff --git a/src/utils/pool_signal.c b/src/utils/pool_signal.c
index 3a7a949cb..57ee3bd4b 100644
--- a/src/utils/pool_signal.c
+++ b/src/utils/pool_signal.c
@@ -165,8 +165,11 @@ int pool_signal_parent(int sig)
* This is a guard against sending the signal to init process
* pgpool-II parent process crashed and left the child processes
* orphan.
+ * we make a little exception for pcp process children, since their
+ * they want to signal the main pgpool-II process but main process is
+ * not the direct parent
*/
- if(mypid != getppid())
+ if(processType != PT_PCP_WORKER && mypid != getppid())
{
/*
* pgpool parent is no more alive, commiting sucide.
diff --git a/src/watchdog/test/stab.c b/src/watchdog/test/stab.c
index 1fd7241c8..ca017a128 100644
--- a/src/watchdog/test/stab.c
+++ b/src/watchdog/test/stab.c
@@ -20,16 +20,11 @@ static void child_wait(int signo);
void pool_error(const char *fmt,...)
{
va_list ap;
+ pool_sigset_t oldmask;
#ifdef HAVE_ASPRINTF
char *fmt2;
int len;
#endif
-
-#ifdef HAVE_SIGPROCMASK
- sigset_t oldmask;
-#else
- int oldmask;
-#endif
/* Write error message to syslog */
if (pool_config->logsyslog == 1) {
va_start(ap, fmt);
[parent: 9ca5ebd6db8f]