/*--------------------------------------------------------------------
 * FILE:
 *     child.c
 *
 * NOTE:
 *     This file is composed of the functions to call with the source
 *     at child process of pglb.
 *
 * Portions Copyright (c) 2003-2006, Atsushi Mitani
 * Portions Copyright (c) 2003-2006, Tatsuo Ishii
 *--------------------------------------------------------------------
 */
/*
 * 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.
 *
*/
#include "postgres.h"

#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <sys/wait.h>
#include <sys/time.h>
#include <ctype.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/ipc.h>
#include <netdb.h>
#include <netinet/in.h>
#include <errno.h>
#include <fcntl.h>
#include <time.h>
#include <sys/param.h>
#include <arpa/inet.h>
#include <sys/file.h>

#ifdef HAVE_NETINET_TCP_H
#include <netinet/tcp.h>
#endif

#ifdef HAVE_CRYPT_H
#include <crypt.h>
#endif

#include "postgres_fe.h"
#include "libpq/pqcomm.h"

#include "replicate_com.h"
#include "pglb.h"

/*--------------------------------------
 * GLOBAL VARIABLE DECLARATION
 *--------------------------------------
 */
POOL_CONNECTION * Frontend = NULL;

/*--------------------------------------
 * PROTOTYPE DECLARATION
 *--------------------------------------
 */
int PGRpre_fork_children(ClusterTbl * ptr);
int PGRpre_fork_child(ClusterTbl * ptr);
int PGRdo_child( int use_pool);
int PGRcreate_child(ClusterTbl * cluster_p);
pid_t PGRscan_child_tbl(ClusterTbl * cluster_p);
void notice_backend_error(void);
void do_pooling_child(int sig);
int PGRset_status_to_child_tbl(pid_t pid, int status);
int PGRadd_child_tbl(ClusterTbl * cluster_p, pid_t pid, int status);
int PGRget_child_status(pid_t pid);
void PGRreturn_connection_full_error(void);
void PGRreturn_no_connection_error(void);
void PGRquit_children_on_cluster(int rec_no);

#ifdef NONE_BLOCK
static void set_nonblock(int fd);
#endif
static void unset_nonblock(int fd);
static POOL_CONNECTION *do_accept(int unix_fd, int inet_fd);
static PGR_StartupPacket *read_startup_packet(POOL_CONNECTION *cp);
static int send_startup_packet(POOL_CONNECTION_POOL_SLOT *cp);
static void cancel_request(CancelPacket *sp, int secondary_backend);
static POOL_CONNECTION_POOL *connect_backend(PGR_StartupPacket *sp, POOL_CONNECTION *frontend);
static int send_params(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend);
static void child_end(int sig);
static void PGRreturn_with_error(char *msg);


/*--------------------------------------------------------------------
 * SYMBOL
 *    PGRpre_fork_children()
 * NOTES
 *    pre forked child precesses
 * ARGS
 *    ClusterTbl * ptr: pointer of cluster server table (I)
 * RETURN
 *    OK: STATUS_OK
 *    NG: STATUS_ERROR
 *--------------------------------------------------------------------
 */
int
PGRpre_fork_children(ClusterTbl * ptr)
{
	int cnt;

	if (ptr == NULL)
	{
		return STATUS_ERROR;
	}
	cnt = 0 ;
	while ((ptr->useFlag != TBL_END) && (cnt < ClusterNum))
	{
		PGRpre_fork_child(ptr);
		cnt ++;
		ptr ++;
	}
	return STATUS_OK;
}

/*--------------------------------------------------------------------
 * SYMBOL
 *    PGRpre_fork_child()
 * NOTES
 *    pre forked child precess
 * ARGS
 *    ClusterTbl * ptr: pointer of cluster server table (I)
 * RETURN
 *    OK: STATUS_OK
 *    NG: STATUS_ERROR
 *--------------------------------------------------------------------
 */
int
PGRpre_fork_child(ClusterTbl * ptr)
{
	pid_t pid = 0;
	int i;

	if (ptr == NULL)
	{
		return STATUS_ERROR;
	}
	if (ptr->useFlag == TBL_END)
	{
		return STATUS_ERROR;
	}
	for ( i = 0 ; i < ptr->max_connect * Max_Pool ; i ++)
	{
		pid = PGRcreate_child(ptr);		
	}
	return STATUS_OK;
}
/*--------------------------------------------------------------------
 * SYMBOL
 *    PGRdo_child()
 * NOTES
 *    execute child process
 * ARGS
 *    int use_pool: usage flag of connection pooling (I)
 * RETURN
 *    OK: STATUS_OK
 *    NG: STATUS_ERROR
 *--------------------------------------------------------------------
 */
int
PGRdo_child( int use_pool)
{
	char * func = "PGRdo_child()";
	pid_t pid = 0;
	PGR_StartupPacket *sp = NULL;
	POOL_CONNECTION *frontend = NULL;
	POOL_CONNECTION_POOL *backend = NULL;
	int status = 0;
	int connection_reuse = 1;
	int ssl_request = 0;
	int count = 0;

	pid = getpid();
#ifdef PRINT_DEBUG
	show_debug("%s:I am %d",func, pid);
#endif			

	/* set up signal handlers */
	signal(SIGALRM, SIG_DFL);
	signal(SIGTERM, child_end);
	signal(SIGHUP, child_end);
	signal(SIGINT, child_end);
	signal(SIGUSR1, SIG_IGN);
	signal(SIGUSR2, SIG_IGN);

#ifdef NONE_BLOCK
	/* set listen fds to none block */
	set_nonblock(Frontend_FD.unix_fd);
	set_nonblock(Frontend_FD.inet_fd);
#endif

retry_accept:
	/* perform accept() */
	frontend = do_accept(Frontend_FD.unix_fd,Frontend_FD.inet_fd);
	if (frontend == NULL)
	{
		/* accept() failed. return to the accept() loop */
		PGRset_status_to_child_tbl(pid,TBL_FREE);
		return STATUS_ERROR;
	}

	/* unset frontend fd tp none block */
	unset_nonblock(frontend->fd);

	/* read the startup packet */
	sp = 0;
retry_startup:
	if (sp)
	{
		free(sp->startup_packet);
		free(sp->database);
		free(sp->user);
		free(sp);
	}

	sp = read_startup_packet(frontend);
	if (sp == NULL)
	{
		/* failed to read the startup packet. return to the
		   accept() loop */
		pool_close(frontend);
		PGRset_status_to_child_tbl(pid,TBL_FREE);
		return STATUS_ERROR;
	}
	PGRset_status_to_child_tbl(pid,TBL_ACCEPT);

	/* cancel request? */
	if (sp->major == 1234 && sp->minor == 5678)
	{
		cancel_request((CancelPacket *)sp->startup_packet, 0);
		pool_close(frontend);
		return STATUS_ERROR;
	}

	/* SSL? */
	if (sp->major == 1234 && sp->minor == 5679)
	{
		/* SSL not supported */
#ifdef PRINT_DEBUG
		show_debug("%s:SSLRequest: sent N; retry startup",func);
#endif			
		if (ssl_request)
		{
			pool_close(frontend);
			return STATUS_ERROR;
		}

		/*
		 * say to the frontend "we do not suppport SSL"
		 * note that this is not a NOTICE response despite it's an 'N'!
		 */
		pool_write_and_flush(frontend, "N", 1);
		ssl_request = 1;
		goto retry_startup;
	}

	/*
	 * Ok, negotiaton with frontend has been done. Let's go to the next step.
	 */
	/*
	 * if there's no connection associated with user and database,
	 * we need to connect to the backend and send the startup packet.
	 */
	count = 0;
	if ((backend = pool_get_cp(sp->user, sp->database, sp->major)) == NULL)
	{
		connection_reuse = 0;

		if ((backend = connect_backend(sp, frontend)) == NULL)
		{
			/*
			PGRset_status_on_cluster_tbl(TBL_ERROR,CurrentCluster);
			return STATUS_ERROR;
			*/
			goto retry_accept;
		}
	}
	else
	{
		/* reuse existing connection to backend */

		if (pool_do_reauth(frontend, backend))
		{
			pool_close(frontend);
			return STATUS_ERROR;
		}

		if (MAJOR(backend) == 3)
		{
			if (send_params(frontend, backend))
			{
				pool_close(frontend);
				return STATUS_ERROR;
			}
		}

		/* send ReadyForQuery to frontend */
		pool_write(frontend, "Z", 1);

		if (MAJOR(backend) == 3)
		{
			int len;
			char tstate;

			len = htonl(5);
			pool_write(frontend, &len, sizeof(len));
			tstate = TSTATE(backend);
			pool_write(frontend, &tstate, 1);
		}

		if (pool_flush(frontend) < 0)
		{
			pool_close(frontend);
			return STATUS_ERROR;
		}

	}

	/* query process loop */
	for (;;)
	{
		POOL_STATUS status;

		status = pool_process_query(frontend, backend, 0);

		switch (status)
		{
			/* client exits */
			case POOL_END:
				/* do not cache connection to template0, template1, regression */
				if (!strcmp(sp->database, "template0") || !strcmp(sp->database, "template1") ||
				    !strcmp(sp->database, "regression") || use_pool == NOT_USE_CONNECTION_POOL)
				{
					pool_close(frontend);
					pool_send_frontend_exits(backend);
					pool_discard_cp(sp->user, sp->database, sp->major);
				}
				else
				{
					POOL_STATUS status1;

					/* send reset request to backend */
					status1 = pool_process_query(frontend, backend, 1);
					pool_close(frontend);

					/* if we detect errors on resetting connection, we need to discard
					 * this connection since it might be in unknown status
					 */
					if (status1 != POOL_CONTINUE)
						pool_discard_cp(sp->user, sp->database, sp->major);
					else
						pool_connection_pool_timer(backend);
				}
				break;
			
			/* error occured. discard backend connection pool
			   and disconnect connection to the frontend */
			case POOL_ERROR:
				show_error("%s:do_child: exits with status 1 due to error",func);
				break;

			/* fatal error occured. just exit myself... */
			case POOL_FATAL:
				show_error("%s:do_child: fatal error occured",func);
				notice_backend_error();
				break;

			/* not implemented yet */
			case POOL_IDLE:
				do_accept(Frontend_FD.unix_fd,Frontend_FD.inet_fd);
#ifdef PRINT_DEBUG
				show_debug("%s:accept while idle",func);
#endif			
				break;

			default:
				break;
		}

		if (status != POOL_CONTINUE)
			break;
	}
	if ((status == POOL_ERROR) || 
		(status == POOL_FATAL))
	{
		PGRset_status_to_child_tbl(pid,TBL_FREE);
		return STATUS_ERROR;
	}
	PGRset_status_to_child_tbl(pid,TBL_INIT);
	return STATUS_OK;
}

/*--------------------------------------------------------------------
 * SYMBOL
 *    PGRcreate_child()
 * NOTES
 *    create child process
 * ARGS
 *    ClusterTbl * ptr: pointer of cluster server table (I)
 * RETURN
 *    OK: STATUS_OK
 *    NG: STATUS_ERROR
 *--------------------------------------------------------------------
 */
int 
PGRcreate_child(ClusterTbl * cluster_p)
{
	char * func = "PGRcreate_child()";
	pid_t pid,pgid;

	if (cluster_p == NULL)
		return STATUS_ERROR;

#ifdef PRINT_DEBUG
	show_debug("%s:create child [%d@%s]",func,cluster_p->port,cluster_p->hostName);
#endif			
	signal(SIGCHLD,PGRrecreate_child);
	pgid = getpgid((pid_t)0);
	pid = fork();
	if (pid < 0)
	{
		show_error("%s:fork() failed. (%s)",func,strerror(errno));
		return STATUS_ERROR;
	}
	if (pid == 0)
	{
		CurrentCluster = cluster_p;
		if (pool_init_cp())
		{
			show_error("%s:pool_init_cp failed",func);
			exit(1);
		}
		signal(SIGCHLD,PGRchild_wait);
		signal(SIGTERM, child_end);
		signal(SIGHUP, child_end);
		signal(SIGINT, child_end);
		signal(SIGUSR1,do_pooling_child);
		setpgid((pid_t)0,pgid);
		for (;;)
		{
			pause();
			signal(SIGUSR1,do_pooling_child);
		}
#ifdef PRINT_DEBUG
		show_debug("%s:create child end [%d@%s]",func,cluster_p->port,cluster_p->hostName);
#endif			
		child_end(SIGTERM);
	}
	else
	{
		PGRadd_child_tbl(cluster_p,pid,TBL_INIT);
	}
	return pid;
}

/*--------------------------------------------------------------------
 * SYMBOL
 *    PGRscan_child_tbl()
 * NOTES
 *    get a child process id that is waiting for connection 
 *    with the cluster server
 * ARGS
 *    ClusterTbl * ptr: pointer of cluster server table (I)
 * RETURN
 *    OK: child process id
 *    NG: 0
 *--------------------------------------------------------------------
 */
pid_t
PGRscan_child_tbl(ClusterTbl * cluster_p)
{
	char * func = "PGRscan_child_tbl()";
	ChildTbl * p;

	if ( cluster_p == NULL)
	{
		show_error("%s:Cluster_Tbl is not initialize",func);
		return STATUS_ERROR;
	}
	p = Child_Tbl;
	if ( p == NULL)
	{
		show_error("%s:Child_Tbl is not initialize",func);
		return STATUS_ERROR;
	}
	while(p->useFlag != TBL_END)
	{
		if (p->pid <= 0)
		{
			p++;
			continue;
		}
		if ((p->useFlag == TBL_INIT) &&
			(p->rec_no == cluster_p->rec_no))
		{
			p->useFlag = TBL_USE;
			return (p->pid);
		}
		p++;
	}
	return 0;
}

/* notice backend connection error using SIGUSR2 */
void
notice_backend_error(void)
{
	pid_t pid = getpid();

	PGRset_status_to_child_tbl(pid,TBL_ERROR);
	PGRset_status_on_cluster_tbl(TBL_ERROR_NOTICE,CurrentCluster);

	/*
	kill(parent, SIGUSR2);
	sleep(1);
	*/
}


/*
 * start up pooling child process
 */
void
do_pooling_child(int sig)
{
	char * func = "do_pooling_child()";
	int rtn;
	pid_t pid;

	pid = getpid();
	rtn = PGRdo_child(USE_CONNECTION_POOL);
	PGRrelease_connection(CurrentCluster);
	if (rtn != STATUS_OK)
	{
		show_error("%s:PGRdo_child failed",func);
		child_end(SIGTERM);
	}
	return ;
}

/*
 * set status in child process table
 */
int
PGRset_status_to_child_tbl(pid_t pid, int status)
{
	char * func = "PGRset_status_to_child_tbl()";
	ChildTbl * p;

	p = Child_Tbl;
	if ( p == NULL)
	{
		show_error("%s:Child_Tbl is not initialize",func);
		return STATUS_ERROR;
	}
	while(p->useFlag != TBL_END)
	{
		if (p->pid == pid)
		{
			p->useFlag = status;
			return STATUS_OK;
		}
		p++;
	}
	return STATUS_ERROR;
}

/*
 * add child process data in child process table
 */
int
PGRadd_child_tbl(ClusterTbl * cluster_p, pid_t pid, int status)
{
	char * func = "PGRadd_child_tbl()";
	ChildTbl * p;

	p = Child_Tbl;
	if ( cluster_p == NULL)
	{
		show_error("%s:Cluster_Tbl is not initialize",func);
		return STATUS_ERROR;
	}
	if ( p == NULL)
	{
		show_error("%s:Child_Tbl is not initialize",func);
		return STATUS_ERROR;
	}
	while(p->useFlag != TBL_END)
	{
		if ((p->useFlag == TBL_FREE) ||
			(p->useFlag == TBL_ERROR))
		{
			p->useFlag = status;
			p->rec_no = cluster_p->rec_no;
			p->pid = pid;
			return STATUS_OK;
		}
		p++;
	}
	return STATUS_ERROR;
}

int
PGRget_child_status(pid_t pid)
{
	char * func = "PGRget_child_status()";
	ChildTbl * p;

	p = Child_Tbl;
	if ( p == NULL)
	{
		show_error("%s:Child_Tbl is not initialize",func);
		return STATUS_ERROR;
	}

	while (p->useFlag != TBL_END)
	{
		if (p->pid == pid)
		{
			return p->useFlag;
		}
		p++;
	}
	return STATUS_ERROR;
}

void 
PGRreturn_connection_full_error(void)
{
  PGRreturn_with_error( "Sorry, backend connection is full\n");
}

void 
PGRreturn_no_connection_error(void) {
  PGRreturn_with_error("pglb could not connect to server: no cluster available.\n");
}

static void 
PGRreturn_with_error (char *msg) 
{
	PGR_StartupPacket *sp = NULL;
	POOL_CONNECTION *frontend = NULL;


	/* perform accept() */
	frontend = do_accept(Frontend_FD.unix_fd,Frontend_FD.inet_fd);
	if (frontend == NULL)
	{
		/* accept() failed. return to the accept() loop */
		return ;
	}
	sp = read_startup_packet(frontend);
	if (sp == NULL)
	{
		/* failed to read the startup packet. return to the
		   accept() loop */
		pool_close(frontend);
		return ;
	}
	pool_write_and_flush(frontend, "E", 1);
	pool_write_and_flush(frontend, msg, strlen(msg)+1);
	pool_close(frontend);
	return ;
}

void
PGRquit_children_on_cluster(int rec_no)
{
	char * func = "PGRquit_children_on_cluster()";
	ChildTbl * p;

	if (Child_Tbl == NULL)
	{
		return;
	}
	signal(SIGCHLD,SIG_IGN);
	p = Child_Tbl;
	while(p->useFlag != TBL_END)
	{
		if (p->rec_no == rec_no) 
		{
			if (kill (p->pid,SIGTERM) == -1)
			{
				show_error("%s:could not stop pid: %d (%s)",func,p->pid,strerror(errno));
				return;
			}
			PGRchild_wait(SIGTERM);
			p->useFlag = DATA_FREE;
		}
		p++;
	}
	if (Use_Connection_Pool)
	{
		signal(SIGCHLD,PGRrecreate_child);
	}
	else
	{
		signal(SIGCHLD,PGRchild_wait);
	}
}

/* -------------------------------------------------------------------
 * private functions
 * -------------------------------------------------------------------
 */

#ifdef NONE_BLOCK
/*
 * set non-block flag
 */
static void set_nonblock(int fd)
{
	char* func = "set_nonblock()";
	int var;

	/* set fd to none blocking */
	var = fcntl(fd, F_GETFL, 0);
	if (var == -1)
	{
		show_error("%s:fcntl failed. %s", func,strerror(errno));
		child_end(SIGTERM);
	}
	if (fcntl(fd, F_SETFL, var | O_NONBLOCK) == -1)
	{
		show_error("%s:fcntl failed. %s", func,strerror(errno));
		child_end(SIGTERM);
	}
}
#endif

/*
 * unset non-block flag
 */
static void unset_nonblock(int fd)
{
	char * func = "unset_nonblock()";
	int var;

	/* set fd to none blocking */
	var = fcntl(fd, F_GETFL, 0);
	if (var == -1)
	{
		show_error("%s,fcntl failed. %s", func,strerror(errno));
		child_end(SIGTERM);
	}
	if (fcntl(fd, F_SETFL, var & ~O_NONBLOCK) == -1)
	{
		show_error("%s,fcntl failed. %s", func,strerror(errno));
		child_end(SIGTERM);
	}
}


/*
* perform accept() and returns new fd
*/
static POOL_CONNECTION *do_accept(int unix_fd, int inet_fd)
{
	char * func = "do_accept()";
    fd_set	readmask;
    int fds;
	struct sockaddr addr;
	socklen_t addrlen;
	int fd = 0;
	int afd;
	int inet = 0;
	POOL_CONNECTION *cp;
#ifdef ACCEPT_PERFORMANCE
	struct timeval now1, now2;
	static long atime;
	static int cnt;
#endif

	FD_ZERO(&readmask);
	FD_SET(unix_fd, &readmask);
	if (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;

		show_error("%s:select() failed. reason %s",func, strerror(errno));
		return NULL;
	}

	if (fds == 0)
		return NULL;

	if (FD_ISSET(unix_fd, &readmask))
	{
		fd = unix_fd;
	}

	if (FD_ISSET(inet_fd, &readmask))
	{
		fd = inet_fd;
		inet++;
	}

	/*
	 * Note that some SysV systems do not work here. For those
	 * systems, we need some locking mechanism for the fd.
	 */
	addrlen = sizeof(addr);

#ifdef ACCEPT_PERFORMANCE
	gettimeofday(&now1,0);
#endif
	afd = accept(fd, &addr, &addrlen);
	if (afd < 0)
	{
		/*
		 * "Resource temporarily unavailable" (EAGAIN or EWOULDBLOCK)
		 * can be silently ignored.
		 */
		if (errno != EAGAIN && errno != EWOULDBLOCK)
			show_error("%s:accept() failed. reason: %s",func, strerror(errno));
		return NULL;
	}
#ifdef ACCEPT_PERFORMANCE
	gettimeofday(&now2,0);
	atime += (now2.tv_sec - now1.tv_sec)*1000000 + (now2.tv_usec - now1.tv_usec);
	cnt++;
	if (cnt % 100 == 0)
	{
		show_error("%s:cnt: %d atime: %ld",func, cnt, atime);
	}
#endif
#ifdef PRINT_DEBUG
	show_debug("%s:I am %d accept fd %d",func, getpid(), afd);
#endif			

	/* set NODELAY and KEEPALIVE options if INET connection */
	if (inet)
	{
		int on = 1;

		if (setsockopt(afd, IPPROTO_TCP, TCP_NODELAY,
					   (char *) &on,
					   sizeof(on)) < 0)
		{
			show_error("%s:do_accept: setsockopt() failed: %s",func, strerror(errno));
			close(afd);
			return NULL;
		}
		if (setsockopt(afd, SOL_SOCKET, SO_KEEPALIVE,
					   (char *) &on,
					   sizeof(on)) < 0)
		{
			show_error("%s:do_accept: setsockopt() failed: %s", func,strerror(errno));
			close(afd);
			return NULL;
		}
	}

	if ((cp = pool_open(afd)) == NULL)
	{
		close(afd);
		return NULL;
	}
	return cp;
}

/*
* read startup packet
*/
static PGR_StartupPacket *read_startup_packet(POOL_CONNECTION *cp)
{
	char * func = "read_startup_packet()";
	PGR_StartupPacket *sp;
	PGR_StartupPacket_v2 *sp2;
	int protov;
	int len;
	char *p;

	sp = (PGR_StartupPacket *)malloc(sizeof(PGR_StartupPacket));
	if (!sp)
	{
		show_error("%s:read_startup_packet: out of memory",func);
		return NULL;
	}

	/* read startup packet length */
	if (pool_read(cp, &len, sizeof(len)))
	{
		free(sp);
		return NULL;
	}
	len = ntohl(len);
	len -= sizeof(len);

	if (len <= 0)
	{
		show_error("%s:read_startup_packet: incorrect packet length (%d)", func,len);
		free(sp);
		return NULL;
	}

	sp->startup_packet = calloc(len, 1);
	if (!sp->startup_packet)
	{
		show_error("%s:read_startup_packet: out of memory",func);
		free(sp);
		return NULL;
	}

	/* read startup packet */
	if (pool_read(cp, sp->startup_packet, len))
	{
		free(sp);
		return NULL;
	}

	sp->len = len;
	memcpy(&protov, sp->startup_packet, sizeof(protov));
	sp->major = ntohl(protov)>>16;
	sp->minor = ntohl(protov) & 0x0000ffff;
	p = sp->startup_packet;

	switch(sp->major)
	{
		case PROTO_MAJOR_V2: /* V2 */
			sp2 = (PGR_StartupPacket_v2 *)(sp->startup_packet);

			sp->database = calloc(SM_DATABASE+1, 1);
			if (!sp->database)
			{
				show_error("%s:read_startup_packet: out of memory",func);
				free(sp);
				return NULL;
			}
			strncpy(sp->database, sp2->database, SM_DATABASE);

			sp->user = calloc(SM_USER+1, 1);
			if (!sp->user)
			{
				show_error("%s:read_startup_packet: out of memory",func);
				free(sp);
				return NULL;
			}
			strncpy(sp->user, sp2->user, SM_USER);

			break;

		case PROTO_MAJOR_V3: /* V3 */
			p += sizeof(int);	/* skip protocol version info */

			while(*p)
			{
				if (!strcmp("user", p))
				{
					p += (strlen(p) + 1);
					sp->user = strdup(p);
					if (!sp->user)
					{
						show_error("%s:read_startup_packet: out of memory",func);
						free(sp);
						return NULL;
					}
				}
				else if (!strcmp("database", p))
				{
					p += (strlen(p) + 1);
					sp->database = strdup(p);
					if (!sp->database)
					{
						show_error("%s:read_startup_packet: out of memory",func);
						free(sp);
						return NULL;
					}
				}
				p += (strlen(p) + 1);
			}
			break;

		case 1234:		/* cancel or SSL request */
			/* set dummy database, user info */
			sp->database = calloc(1, 1);
			if (!sp->database)
			{
				show_error("%s:read_startup_packet: out of memory",func);
				free(sp);
				return NULL;
			}
			sp->user = calloc(1, 1);
			if (!sp->user)
			{
				show_error("%s:read_startup_packet: out of memory",func);
				free(sp);
				return NULL;
			}
			break;

		default:
			show_error("%s:read_startup_packet: invalid major no: %d",func, sp->major);
			free(sp);
			return NULL;
	}

#ifdef PRINT_DEBUG
	show_debug("%s:Protocol Major: %d Minor: %d database: %s user: %s", 
			   func,sp->major, sp->minor, sp->database, sp->user);
#endif			

	return sp;
}

/*
* send startup packet
*/
static int send_startup_packet(POOL_CONNECTION_POOL_SLOT *cp)
{
	int len;

	len = htonl(cp->sp->len + sizeof(len));
	pool_write(cp->con, &len, sizeof(len)); 
	return pool_write_and_flush(cp->con, cp->sp->startup_packet, cp->sp->len);
}

/*
 * process cancel request
 */
static void cancel_request(CancelPacket *sp, int secondary_backend)
{
	char * func = "cancel_request()";
	int	len;
	int fd;
	POOL_CONNECTION *con;
	char hostName[128];

#ifdef PRINT_DEBUG
	show_debug("%s:Cancel request received",func);
#endif			

	if (CurrentCluster == NULL)
	{
		return;
	}
	if (gethostname(hostName,sizeof(hostName)) < 0)
	{
		show_error("%s:gethostname() failed. (%s)",func,strerror(errno));
		return ;
	}
	if (secondary_backend)
	{
		if (PGRis_same_host(hostName,CurrentCluster->hostName))
			fd = connect_unix_domain_socket(1);
		else
			fd = connect_inet_domain_socket(1);
	}
	else
	{
		if (PGRis_same_host(hostName,CurrentCluster->hostName))
			fd = connect_unix_domain_socket(0);
		else
			fd = connect_inet_domain_socket(0);
	}

	if (fd < 0)
	{
		show_error("%s:Could not create socket for sending cancel request",func);
		return;
	}

	con = pool_open(fd);
	if (con == NULL)
		return;

	len = htonl(sizeof(len) + sizeof(CancelPacket));
	pool_write(con, &len, sizeof(len));

	if (pool_write_and_flush(con, sp, sizeof(CancelPacket)) < 0)
		show_error("%s:Could not send cancel request packet",func);
	pool_close(con);
}

static POOL_CONNECTION_POOL *connect_backend(PGR_StartupPacket *sp, POOL_CONNECTION *frontend)
{
	char * func ="connect_backend()";
	POOL_CONNECTION_POOL *backend;

	/* connect to the backend */
	backend = pool_create_cp();
	if (backend == NULL)
	{
		pool_send_error_message(frontend, sp->major, "XX000", "connection cache is full", "",
								"increace max_pool", __FILE__, __LINE__);
		pool_close(frontend);
		return NULL;
	}

	/* mark this is a backend connection */
	backend->slots[0]->con->isbackend = 1;
	/*
	 * save startup packet info
	 */
	backend->slots[0]->sp = sp;

	if (pool_config_replication_enabled)
	{
		backend->slots[1]->con->isbackend = 1;
		backend->slots[1]->con->issecondary_backend = 1;
		/*
		 * save startup packet info
		 */
		backend->slots[1]->sp = sp;
	}

	/* send startup packet */
	if (send_startup_packet(backend->slots[0]) < 0)
	{
		show_error("%s:do_child: fails to send startup packet to the backend",func);
		pool_close(frontend);
		return NULL;
	}

	/* send startup packet */
	if (pool_config_replication_enabled)
	{
		if (send_startup_packet(backend->slots[1]) < 0)
		{
			show_error("%s:do_child: fails to send startup packet to the secondary backend",func);
			pool_close(frontend);
			return NULL;
		}
	}

	/*
	 * do authentication stuff
	 */
	if (pool_do_auth(frontend, backend))
	{
		pool_close(frontend);
		pool_discard_cp(sp->user, sp->database, sp->major);
		return NULL;
	}
	return backend;
}

static int send_params(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend)
{
	char * func = "send_params()";
	int index;
	char *name, *value;
	int len, sendlen;

	index = 0;
	while (pool_get_param(&MASTER(backend)->params, index++, &name, &value) == 0)
	{
		pool_write(frontend, "S", 1);
		len = sizeof(sendlen) + strlen(name) + 1 + strlen(value) + 1;
		sendlen = htonl(len);
		pool_write(frontend, &sendlen, sizeof(sendlen));
		pool_write(frontend, name, strlen(name) + 1);
		pool_write(frontend, value, strlen(value) + 1);
	}

	if (pool_flush(frontend))
	{
		show_error("%s:pool_send_params: pool_flush() failed",func);
		return -1;
	}
	return 0;
}

/*
 * ending function of child process
 */
static void
child_end(int sig)
{
	signal(sig,SIG_IGN);

	pool_finish();
	exit(0);
}
