/****************************************************************
 *								*
 * Copyright (c) 2006-2023 Fidelity National Information	*
 * Services, Inc. and/or its subsidiaries. All rights reserved.	*
 *								*
 *	This source code contains the intellectual property	*
 *	of its copyright holder(s), and is made available	*
 *	under a license.  If you do not know the terms of	*
 *	the license, please stop and do not read further.	*
 *								*
 ****************************************************************/

#if defined(__MVS__) && !defined(_ISOC99_SOURCE)
#define _ISOC99_SOURCE
#endif

#include "mdef.h"

#include "gtm_socket.h"
#include "gtm_inet.h"
#include "gtm_netdb.h"
#include "gtm_time.h"
#include "gtm_fcntl.h"
#include "gtm_unistd.h"
#include "gtm_string.h"
#include "gtm_stdio.h"	/* for FILE * in repl_comm.h */

#include <sys/time.h>
#include <errno.h>

#include "gdsroot.h"
#include "gdsblk.h"
#include "gtm_facility.h"
#include "fileinfo.h"
#include "gdsbt.h"
#include "gdsfhead.h"
#include "filestruct.h"
#include "gtmrecv.h"
#include "repl_comm.h"
#include "repl_msg.h"
#include "repl_dbg.h"
#include "repl_errno.h"
#include "io.h"
#include "iosp.h"
#include "eintr_wrappers.h"
#include "jnl.h"
#include "repl_sp.h"
#include "repl_filter.h"
#include "repl_log.h"
#include "gtmsource.h"
#include "sgtm_putmsg.h"
#include "gt_timer.h"
#include "min_max.h"
#include "error.h"
#include "copy.h"
#include "repl_instance.h"
#include "ftok_sems.h"
#include "buddy_list.h"		/* needed for muprec.h */
#include "hashtab_mname.h"	/* needed for muprec.h */
#include "hashtab_int4.h"	/* needed for muprec.h */
#include "hashtab_int8.h"	/* needed for muprec.h */
#include "muprec.h"
#include "gtmmsg.h"
#include "is_proc_alive.h"
#include "jnl_typedef.h"
#include "memcoherency.h"
#include "have_crit.h"			/* needed for ZLIB_UNCOMPRESS */
#include "deferred_signal_handler.h"	/* needed for ZLIB_UNCOMPRESS */
#include "gtm_zlib.h"
#include "wbox_test_init.h"
#ifdef GTM_TRIGGER
#include "repl_sort_tr_buff.h"
#endif
#include "replgbl.h"
#include "gtmio.h"
#include "repl_inst_dump.h"		/* for "repl_dump_histinfo" prototype */
#include "gv_trigger_common.h"
#include "anticipatory_freeze.h"
#ifdef GTM_TLS
#include "gtm_repl.h"
#endif
#include "util.h"			/* for OUT_BUFF_SIZE */

#define	GTM_ZLIB_UNCMP_ERR_STR		"error from zlib uncompress function "
#define	GTM_ZLIB_Z_MEM_ERROR_STR	"Out-of-memory " GTM_ZLIB_UNCMP_ERR_STR
#define	GTM_ZLIB_Z_BUF_ERROR_STR	"Insufficient output buffer " GTM_ZLIB_UNCMP_ERR_STR
#define	GTM_ZLIB_Z_DATA_ERROR_STR	"Input-data-incomplete-or-corrupt " GTM_ZLIB_UNCMP_ERR_STR
#define	GTM_ZLIB_UNCMPLEN_ERROR_STR	"Decompressed message data length %d is not equal to precompressed length %d "
#define	GTM_ZLIB_UNCMP_ERR_SEQNO_STR	"at seqno "INT8_FMT" "INT8_FMTX"\n"
#define	GTM_ZLIB_UNCMP_ERR_SOLVE_STR	"before sending REPL_CMP_SOLVE message\n"
#define	GTM_ZLIB_UNCMPTRANSITION_STR	"Defaulting to NO decompression\n"

#define RECVBUFF_REPLMSGLEN_FACTOR 		8

#define GTMRECV_WAIT_FOR_STARTJNLSEQNO		100 /* ms */

#define GTMRECV_WAIT_FOR_UPD_PROGRESS		100 /* ms */
#define PROC_RECVOPS_PRINT_MSG_LEN		2048 /* bytes*/
/* By having different high and low watermarks, we can reduce the # of XOFF/XON exchanges */
#define RECVPOOL_HIGH_WATERMARK_PCTG		90	/* Send XOFF when %age of receive pool space occupied goes beyond this */
#define RECVPOOL_LOW_WATERMARK_PCTG		80	/* Send XON when %age of receive pool space occupied falls below this */
#define RECVPOOL_XON_TRIGGER_SIZE		(1 * 1024 * 1024) /* Keep the low water mark within this amount of high water mark
								   * so that we don't wait too long to send XON */

#define GTMRECV_XOFF_LOG_CNT			100

#define GTMRECV_HEARTBEAT_PERIOD		10	/* seconds, timer that goes off every this period is the time keeper for
							 * receiver server; used to reduce calls to time related systemc calls */

#define ONLN_RLBK_CMD_MAXLEN			1024
#define MUPIP_DIST_STR				"$gtm_dist/mupip "
#define	ONLN_RLBK_CMD				"journal "
#define ONLN_RLBK_VERBOSE			"-verbose "
#define	ONLN_RLBK_QUALIFIERS			"-online -rollback -backward \"*\" -fetchresync=" /* port# will be filled later */

#if defined(__hpux) && !defined(__hppa) || defined(_AIX)
#define KEEPALIVE_PROTO_LEVEL IPPROTO_TCP
#define KEEPALIVE_TIME		5
#define KEEPALIVE_INTVL		5
#define KEEPALIVE_PROBES	5
#elif defined(__linux__)
#define KEEPALIVE_PROTO_LEVEL SOL_TCP
#define KEEPALIVE_TIME		5
#define KEEPALIVE_INTVL		5
#define KEEPALIVE_PROBES	5
#endif

GBLDEF	repl_msg_ptr_t		gtmrecv_msgp;
GBLDEF	int			gtmrecv_max_repl_msglen;
GBLDEF	int			gtmrecv_sock_fd = FD_INVALID;
GBLDEF	boolean_t		repl_connection_reset = TRUE;
GBLDEF	boolean_t		gtmrecv_wait_for_jnl_seqno = FALSE;
GBLDEF	boolean_t		gtmrecv_bad_trans_sent = FALSE;
GBLDEF	boolean_t		gtmrecv_send_cmp2uncmp = FALSE;

GBLDEF	qw_num			repl_recv_data_recvd = 0;
GBLDEF	qw_num			repl_recv_data_processed = 0;
GBLDEF	qw_num			repl_recv_postfltr_data_procd = 0;
GBLDEF	qw_num			repl_recv_lastlog_data_recvd = 0;
GBLDEF	qw_num			repl_recv_lastlog_data_procd = 0;

GBLDEF	time_t			repl_recv_prev_log_time;
GBLDEF	time_t			repl_recv_this_log_time;
GBLDEF	volatile time_t		gtmrecv_now = 0;

STATICDEF uchar_ptr_t		gtmrecv_cmpmsgp;
STATICDEF int			gtmrecv_cur_cmpmsglen;
STATICDEF int			gtmrecv_max_repl_cmpmsglen;
STATICDEF uchar_ptr_t		gtmrecv_uncmpmsgp;
STATICDEF int			gtmrecv_max_repl_uncmpmsglen;
STATICDEF int			gtmrecv_repl_cmpmsglen;
STATICDEF int			gtmrecv_repl_uncmpmsglen;

STATICFNDCL	void	gtmrecv_repl_send_loop_error(int status, char *msgtypestr);
STATICFNDCL	int	repl_tr_endian_convert(unsigned char remote_jnl_ver, uchar_ptr_t jnl_buff, uint4 jnl_len);
STATICFNDCL	void	do_flow_control(gtm_uint64_t write_pos);
STATICFNDCL	int	gtmrecv_est_conn(void);
STATICFNDCL	int	gtmrecv_start_onln_rlbk(void);
STATICFNDCL	void	prepare_recvpool_for_write(gtm_uint64_t datalen, gtm_uint64_t pre_filter_write_len);
STATICFNDCL	void	copy_to_recvpool(uchar_ptr_t databuff, gtm_uint64_t datalen);
STATICFNDCL	void	wait_for_updproc_to_clear_backlog(void);
STATICFNDCL	void	process_tr_buff(int msg_type);
STATICFNDCL	void	gtmrecv_updresync_histinfo_find_seqno(seq_num input_seqno, int4 strm_num, repl_histinfo *histinfo);
STATICFNDCL	void	gtmrecv_updresync_histinfo_get(int4 index, repl_histinfo *histinfo);
STATICFNDCL	void	gtmrecv_process_need_strminfo_msg(repl_needstrminfo_msg_ptr_t need_strminfo_msg);
STATICFNDCL	void	gtmrecv_process_need_histinfo_msg(repl_needhistinfo_msg_ptr_t need_histinfo_msg, repl_histinfo *histinfo);
STATICFNDCL	void	do_main_loop(boolean_t crash_restart);
STATICFNDCL	void	gtmrecv_heartbeat_timer(TID tid, int4 interval_len, int *interval_ptr);
STATICFNDCL	void	gtmrecv_main_loop(boolean_t crash_restart);
#ifdef GTM_TLS
STATICFNDCL	boolean_t gtmrecv_exchange_tls_info(uint4 remote_API_ver, uint4 remote_lib_ver);
#endif

GBLREF	gtmrecv_options_t	gtmrecv_options;
GBLREF	int			gtmrecv_listen_sock_fd;
GBLREF	recvpool_addrs		recvpool;
GBLREF	boolean_t		gtmrecv_logstats;
GBLREF	int			gtmrecv_filter;
GBLREF	int			gtmrecv_log_fd;
GBLREF	FILE			*gtmrecv_log_fp;
GBLREF	seq_num			seq_num_zero, seq_num_one, seq_num_minus_one;
GBLREF	unsigned char		*repl_filter_buff;
GBLREF	int			repl_filter_bufsiz;
GBLREF	unsigned int		jnl_source_datalen, jnl_dest_maxdatalen;
GBLREF	unsigned char		jnl_source_rectype, jnl_dest_maxrectype;
GBLREF	int			repl_max_send_buffsize, repl_max_recv_buffsize;
GBLREF	seq_num			lastlog_seqno;
GBLREF	uint4			log_interval;
GBLREF	qw_num			trans_recvd_cnt, last_log_tr_recvd_cnt;
GBLREF	jnlpool_addrs_ptr_t	jnlpool;
GBLREF	jnl_gbls_t		jgbl;
GBLREF	mur_opt_struct		mur_options;
GBLREF	mur_gbls_t		murgbl;
GBLREF	repl_conn_info_t	*this_side, *remote_side;
GBLREF	int4			strm_index;

error_def(ERR_INSNOTJOINED);
error_def(ERR_INSROLECHANGE);
error_def(ERR_INSUNKNOWN);
error_def(ERR_JNLNEWREC);
error_def(ERR_JNLSETDATA2LONG);
error_def(ERR_NOSUPPLSUPPL);
error_def(ERR_PRIMARYNOTROOT);
error_def(ERR_RCVRMANYSTRMS);
error_def(ERR_REPL2OLD);
error_def(ERR_REPLCOMM);
error_def(ERR_REPLINSTNOHIST);
error_def(ERR_REPLINSTREAD);
error_def(ERR_REPLNOTLS);
error_def(ERR_REPLTRANS2BIG);
error_def(ERR_REPLXENDIANFAIL);
error_def(ERR_RESUMESTRMNUM);
error_def(ERR_REUSEINSTNAME);
error_def(ERR_REPLAHEAD);
error_def(ERR_STRMNUMIS);
error_def(ERR_SUPRCVRNEEDSSUPSRC);
error_def(ERR_SYSCALL);
error_def(ERR_TEXT);
error_def(ERR_TLSCONVSOCK);
error_def(ERR_TLSHANDSHAKE);
error_def(ERR_UNIMPLOP);
error_def(ERR_UPDSYNCINSTFILE);

typedef enum
{
	GTM_RECV_POOL,
	GTM_RECV_CMPBUFF
} gtmrecv_buff_t;

static	unsigned char	*buffp, *buff_start, *msgbuff;
static	gtm_uint64_t	buff_unprocessed;
static	gtm_uint64_t	buffered_data_len;
static	gtm_uint64_t	max_recv_bufsiz;
static	gtm_uint64_t	data_len;
static	gtm_uint64_t	exp_data_len;
static	boolean_t	xoff_sent;
static	repl_msg_t	xon_msg, xoff_msg;
static	int		xoff_msg_log_cnt = 0;
static	long		recvpool_high_watermark, recvpool_low_watermark;
static	gtm_uint64_t	write_loc, write_wrap;
static	gtm_uint64_t	write_off;
static	double		time_elapsed;
static	gtm_uint64_t	recvpool_size;
static	int		heartbeat_period;
#ifdef REPL_CMP_SOLVE_TESTING
static	boolean_t	repl_cmp_solve_timer_set;
#endif

#define ISSUE_REPLCOMM_ERROR(REASON, SAVE_ERRNO)									\
{															\
	SEND_SYSMSG_REPLCOMM(LEN_AND_LIT(REASON));									\
	if (0 != SAVE_ERRNO)												\
		rts_error_csa(CSA_ARG(NULL) VARLSTCNT(7) ERR_REPLCOMM, 0, ERR_TEXT, 2, LEN_AND_LIT(REASON), SAVE_ERRNO);\
	else														\
		rts_error_csa(CSA_ARG(NULL) VARLSTCNT(6) ERR_REPLCOMM, 0, ERR_TEXT, 2, LEN_AND_LIT(REASON));		\
}

#define	GTMRECV_EXPAND_CMPBUFF_IF_NEEDED(cmpmsglen)			\
{									\
	int	lclcmpmsglen;						\
									\
	lclcmpmsglen = ROUND_UP2(cmpmsglen, REPL_MSG_ALIGN);		\
	if (lclcmpmsglen > gtmrecv_max_repl_cmpmsglen)			\
	{								\
		if (NULL != gtmrecv_cmpmsgp)				\
			free(gtmrecv_cmpmsgp);				\
		gtmrecv_cmpmsgp = (uchar_ptr_t)malloc(lclcmpmsglen);	\
		gtmrecv_max_repl_cmpmsglen = (lclcmpmsglen);		\
	}								\
	assert(0 == (gtmrecv_max_repl_cmpmsglen % REPL_MSG_ALIGN));	\
}

#define	GTMRECV_EXPAND_UNCMPBUFF_IF_NEEDED(uncmpmsglen)			\
{									\
	if (uncmpmsglen > gtmrecv_max_repl_uncmpmsglen)			\
	{								\
		if (NULL != gtmrecv_uncmpmsgp)				\
			free(gtmrecv_uncmpmsgp);			\
		gtmrecv_uncmpmsgp = (uchar_ptr_t)malloc(uncmpmsglen);	\
		gtmrecv_max_repl_uncmpmsglen = (uncmpmsglen);		\
	}								\
	assert(0 == (gtmrecv_max_repl_uncmpmsglen % REPL_MSG_ALIGN));	\
}

/* Set an alternate buffer as the target to hold the incoming compressed journal records.
 * This will then be uncompressed into yet another buffer from where the records will be
 * transferred to the receive pool one transaction at a time.
 */
#define	GTMRECV_SET_BUFF_TARGET_CMPBUFF(cmplen, uncmplen, curcmplen)	\
{									\
	GTMRECV_EXPAND_CMPBUFF_IF_NEEDED(cmplen);			\
	GTMRECV_EXPAND_UNCMPBUFF_IF_NEEDED(uncmplen);			\
	curcmplen = 0;							\
}

/* Wrap the "prepare_recvpool_for_write", "copy_to_recvpool", "do_flow_control" and "gtmrecv_poll_actions" functions in macros.
 * This is needed because every invocation of these functions (except for a few gtmrecv_poll_actions invocations) should check
 * a few global variables to determine if there was an error and in that case return from the caller.
 * All callers of these functions should use the macro and not invoke the function directly.
 * This avoids duplication of the error check logic.
 */
#define	PREPARE_RECVPOOL_FOR_WRITE(curdatalen, prefilterdatalen)		\
{										\
	prepare_recvpool_for_write(curdatalen, prefilterdatalen);		\
		/* could update "recvpool_ctl->write" and "write_loc" */	\
	if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)		\
		return;								\
}
#define	COPY_TO_RECVPOOL(ptr, datalen)						\
{										\
	copy_to_recvpool(ptr, datalen); /* uses and updates "write_loc" */	\
	if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)		\
		return;								\
}

#if defined(GTM_REPLIC_FLOW_CONTROL_ENABLED)
#define	DO_FLOW_CONTROL(write_loc)						\
{										\
	do_flow_control(write_loc);						\
	if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)		\
		return;								\
}
#else
#define	DO_FLOW_CONTROL(write_loc)	/* nothing */
#endif

/* Wrapper for gtmrecv_poll_actions to handle connection reset. The arguments passed in are typically either the static variables
 * `data_len', `buff_unprocessed' and `buffp'. But, these values are relevant only when this is being called from do_main_loop as
 * they indicate how much of the received buffer is still to be processed and the pointer to the beginning of the unprocessed
 * buffer. Outside do_main_loop (for instance, in gtmrecv_est_conn), gtmrecv_poll_actions is invoked with 0,0, NULL respectively.
 */
#define	GTMRECV_POLL_ACTIONS(data_len, buff_unprocessed, buffp)			\
{										\
	gtmrecv_poll_actions(data_len, buff_unprocessed, buffp);		\
	if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)		\
		return;								\
}
#define WACKY_MESSAGE(MSG, SOCK_FD, TYPE)								\
MBSTART {	/* If the header is wacky, assume it's a rogue transmission and reset the connection */	\
	repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received UNKNOWN message (type = %d / %d). "		\
	"Discarding it and resetting connection.\n", MSG & REPL_TR_CMP_MSG_TYPE_MASK, TYPE);		\
	repl_connection_reset = TRUE;									\
	repl_close(SOCK_FD);										\
	return;												\
} MBEND

/* The below macro is used (within this module) to check for errors (and issue appropriate message) after REPL_SEND_LOOP
 * returns.
 * NOTE: This macro, in its current form, cannot be used in all REPL_SEND_LOOP usages due to specific post-error processing
 * done in a few places. There is scope to:
 * (a) Improve the macro to adjust to post-error processing
 * (b) Provide similar macro to be used for REPL_RECV_LOOP usages
 */
#define	CHECK_REPL_SEND_LOOP_ERROR(status, msgstr)		\
{								\
	if (SS_NORMAL != status)				\
	{							\
		gtmrecv_repl_send_loop_error(status, msgstr);	\
		if (repl_connection_reset)			\
			return;					\
	}							\
}

#define GTMRECV_ONLN_RLBK_CLNUP_IF_NEEDED					\
{										\
	sgmnt_addrs	*repl_csa;						\
										\
	repl_csa = &FILE_INFO(jnlpool->jnlpool_dummy_reg)->s_addrs;		\
	assert(repl_csa->now_crit);						\
	assert(NULL != jnlpool->jnlpool_ctl);					\
	if (repl_csa->onln_rlbk_cycle != jnlpool->jnlpool_ctl->onln_rlbk_cycle)	\
	{									\
		SYNC_ONLN_RLBK_CYCLES;						\
		rel_lock(jnlpool->jnlpool_dummy_reg);				\
		gtmrecv_onln_rlbk_clnup();					\
		if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)	\
			return;							\
	}									\
}

/* For cross-endian conversion to happen on the receiving side, the receiver must understand the layout of the journal
 * records. To keep the endian conversion logic on both primary and secondary simple, the following scheme is used:
 * (a) If primary < secondary, endian conversion will happen on primary.
 * (b) If primary >= secondary, primary will apply internal filters to convert the records to secondary's format. The
 *     secondary on receiving them will do the necessary endian conversion before letting the update process see them.
 *
 * However, the above logic will cause the older versions (< V5.4-002) to NOT replicate to V5.4-002 as the endian-conversion
 * by-primary is introduced only from V5.4-002 and above. Hence, allow secondary to do endian conversion for this special
 * case when the primary is a GT.M version running V5.3-003 (V18_JNL_VER) to V5.4-001 (V20_JNL_VER). The lower limit is
 * chosen to be V5.3-003 since that was the first version where cross-endian conversion was supported.
 *
 * There is one other exception. V5.5 source server (V22_JNL_VER) had a bug wherein a history record is endian-converted
 * when the replication is NOT cross-endian and vice versa. In either case, do an endian conversion of the history record.
 *
 * The below macro takes all the above conditions into consideration to determine if the receiver server needs to do endian
 * converison or not.
 */
#define ENDIAN_CONVERSION_NEEDED(IS_NEW_HISTREC, THIS_JNL_VER, REMOTE_JNL_VER, X_ENDIAN)			\
	((IS_NEW_HISTREC && (V22_JNL_VER == REMOTE_JNL_VER)) 							\
		|| (X_ENDIAN && ((REMOTE_JNL_VER >= THIS_JNL_VER) || (V21_JNL_VER > REMOTE_JNL_VER))))

STATICFNDEF void gtmrecv_repl_send_loop_error(int status, char *msgtypestr)
{
	char		print_msg[1024];

	assert((EREPL_SEND == repl_errno) || (EREPL_SELECT == repl_errno));
	if (REPL_CONN_RESET(status) && EREPL_SEND == repl_errno)
	{
		SNPRINTF(print_msg, SIZEOF(print_msg), "Connection got reset while sending %s message. Status = %d ; %s\n",
				msgtypestr, status, STRERROR(status));
		rts_error_csa(CSA_ARG(NULL) VARLSTCNT(6) MAKE_MSG_INFO(ERR_REPLCOMM), 0, ERR_TEXT, 2,
				LEN_AND_STR(print_msg));
		repl_connection_reset = TRUE;
		repl_close(&gtmrecv_sock_fd);
		SNPRINTF(print_msg, SIZEOF(print_msg), "Closing connection on receiver side\n");
		repl_log(gtmrecv_log_fp, TRUE, TRUE, print_msg);
		return;
	} else if (EREPL_SEND == repl_errno)
	{
		SNPRINTF(print_msg, SIZEOF(print_msg), "Error sending %s message. Error in send : %s",
				msgtypestr, STRERROR(status));
		RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(6) ERR_REPLCOMM, 0, ERR_TEXT, 2, LEN_AND_STR(print_msg));
	} else if (EREPL_SELECT == repl_errno)
	{
		SNPRINTF(print_msg, SIZEOF(print_msg), "Error sending %s message. Error in select : %s",
				msgtypestr, STRERROR(status));
		RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(6) ERR_REPLCOMM, 0, ERR_TEXT, 2, LEN_AND_STR(print_msg));
	}
}

/* convert endianness of transaction */
STATICFNDEF int repl_tr_endian_convert(unsigned char remote_jnl_ver, uchar_ptr_t jnl_buff, uint4 jnl_len)
{
	unsigned char			*jb, *jstart, *ptr;
	enum jnl_record_type		rectype;
	int				status, reclen;
	uint4				jlen;
	jrec_prefix			*prefix;
	jnl_record			*rec;
	jnl_string			*keystr;
	mstr_len_t			*vallen_ptr;
	mstr_len_t			temp_val;
	repl_old_triple_jnl_ptr_t	oldtriple;
	repl_histinfo			*histinfo;
	jnl_str_len_t			nodeflags_keylen;
	uint4				update_num, num_participants_4bytes;
	unsigned short			num_participants_2bytes;
#	ifdef DEBUG
	DCL_THREADGBL_ACCESS;

	SETUP_THREADGBL_ACCESS;
#	endif
	assert((remote_jnl_ver >= this_side->jnl_ver) || (V21_JNL_VER > remote_jnl_ver) || (V22_JNL_VER == remote_jnl_ver));
	jb = jnl_buff;
	status = SS_NORMAL;
	jlen = jnl_len;
	assert(0 == ((UINTPTR_T)jb % SIZEOF(UINTPTR_T)));
	while (JREC_PREFIX_SIZE <= jlen)
	{
		assert(0 == ((UINTPTR_T)jb % SIZEOF(UINTPTR_T)));
		rec = (jnl_record *) jb;
		rectype = (enum jnl_record_type)rec->prefix.jrec_type;
		reclen = rec->prefix.forwptr = GTM_BYTESWAP_24(rec->prefix.forwptr);
		if (!IS_REPLICATED(rectype) || (0 == reclen) || (reclen > jlen))
		{	/* Bad OR Incomplete record */
			assert(FALSE);
			status = -1;
			break;
		}
		assert(!IS_ZTP(rectype));
		assert((JRT_HISTREC == rectype) || (JRT_TRIPLE == rectype) || IS_SET_KILL_ZKILL_ZTWORM_LGTRIG_ZTRIG(rectype)
		       || (JRT_TCOM == rectype) || (JRT_NULL == rectype));
		DEBUG_ONLY(jstart = jb;)
		if (JRT_HISTREC == rectype)
		{
			histinfo = &(((repl_histrec_jnl_ptr_t)rec)->histcontent);
			ENDIAN_CONVERT_REPL_HISTINFO(histinfo);
		} else if (JRT_TRIPLE == rectype)
		{
			oldtriple = (repl_old_triple_jnl_ptr_t)rec;
			oldtriple->cycle = GTM_BYTESWAP_32(oldtriple->cycle);
			oldtriple->start_seqno = GTM_BYTESWAP_64(oldtriple->start_seqno);
		} else
		{	/* pini_addr, time, checksum and tn field of the journal records created by the source server are
			 * irrelevant to the receiver server and hence no point doing the endian conversion for them.
			 */
			((jrec_suffix *)((unsigned char *)rec + reclen - JREC_SUFFIX_SIZE))->backptr = reclen;
			rec->jrec_null.jnl_seqno = GTM_BYTESWAP_64(rec->jrec_null.jnl_seqno);
			/* Starting jnl ver V22, we have a "strm_seqno" field in the journal record so endian convert that */
			if (V22_JNL_VER <= remote_jnl_ver)
			{	/* At this point, we could have a TCOM or NULL or SET/KILL/ZKILL/ZTRIG type of record.
				 * Assert that all of them have "strm_seqno" at the exact same offset so we can avoid
				 * an if/then/else check on the record types in order to endian convert "strm_seqno".
				 */
				assert(&rec->jrec_null.strm_seqno == &rec->jrec_set_kill.strm_seqno);
				assert(&rec->jrec_null.strm_seqno == &rec->jrec_tcom.strm_seqno);
				rec->jrec_null.strm_seqno = GTM_BYTESWAP_64(rec->jrec_null.strm_seqno);
			}
			if (IS_SET_KILL_ZKILL_ZTWORM_LGTRIG_ZTRIG(rectype))
			{	/* This code will need changes in case the jnl-ver changes from V28 to V29 so add an assert to
				 * alert to that possibility. Once the code is fixed for the new jnl format, change the assert
				 * to reflect the new latest jnl-ver.
				 */
				assert(JNL_VER_THIS == V28_JNL_VER);
				/* To better understand the logic below (particularly the use of hardcoded offsets), see comment
				 * in repl_filter.c (search for "struct_jrec_upd layout" for the various jnl versions we support).
				 */
				if (V22_JNL_VER <= remote_jnl_ver)
				{	/* byte-swap update_num */
					assert(&rec->jrec_set_kill.update_num == &rec->jrec_ztworm.update_num);
					assert(&rec->jrec_set_kill.update_num == &rec->jrec_lgtrig.update_num);
					rec->jrec_set_kill.update_num = GTM_BYTESWAP_32(rec->jrec_set_kill.update_num);
					/* No need to byte-swap num_participants as it is not used by the update process */
					/* Get pointer to mumps_node */
					keystr = (jnl_string *)&rec->jrec_set_kill.mumps_node;
					assert(keystr == (jnl_string *)&rec->jrec_ztworm.ztworm_str);
					assert(keystr == (jnl_string *)&rec->jrec_lgtrig.lgtrig_str);
				} else if (V19_JNL_VER <= remote_jnl_ver)
				{	/* byte-swap update_num */
					ptr = (unsigned char *)rec + 32; /* is offset of update_num in V19 struct_jrec_upd */
					update_num = *(uint4 *)ptr;
					*(uint4 *)ptr = GTM_BYTESWAP_32(update_num);
					/* No need to byte-swap num_participants as it is not used by the update process */
					/* Get pointer to mumps_node */
					keystr = (jnl_string *)((unsigned char *)rec + 40); /* is offset of mumps_node */
				} else
				{
					assert(V17_JNL_VER <= remote_jnl_ver);
					/* Note: In V17, there is no update_num or num_participants like V19 so no endian convert */
					/* Get pointer to mumps_node */
					keystr = (jnl_string *)((unsigned char *)rec + 32); /* is offset of mumps_node */
				}
				/* In V18, the jnl_string contained a 32 bit length field followed by mumps_node
				 * In V19, the "length" field is divided into 8 bit "nodeflags" and 24 bit "length" fields.
				 * Byteswap the entire 32 bit value
				 */
				nodeflags_keylen = *(jnl_str_len_t *)keystr;
				*(jnl_str_len_t *)keystr = GTM_BYTESWAP_32(nodeflags_keylen);
				if (IS_SET(rectype))
				{
					assert(!IS_ZTWORM(rectype));
					assert(!IS_LGTRIG(rectype));
					/* SET records have a 'value' part which needs to be endian converted */
					vallen_ptr = (mstr_len_t *)&keystr->text[keystr->length];
					GET_MSTR_LEN(temp_val, vallen_ptr);
					temp_val = GTM_BYTESWAP_32(temp_val);
					PUT_MSTR_LEN(vallen_ptr, temp_val);
					/* The actual 'value' itself is a character array and hence needs no endian conversion */
				}
			} else if (JRT_TCOM == rectype)
			{	/* byte-swap num_participants as this is relied upon by "repl_sort_tr_buff".
				 * The offset and size of this field are different for older versions. Endian convert accordingly.
 				 *       V22 struct_jrec_tcom layout is as follows.
				 *      	offset = 0042 [0x002a]      size = 0002 [0x0002]    ----> num_participants
 				 *       V19/V20/V21 struct_jrec_tcom layout is as follows.
 				 *      	offset = 0034 [0x0022]      size = 0002 [0x0002]    ----> num_participants
				 *       V17/V18 struct_jrec_tcom layout is as follows.
				 *      	offset = 0040 [0x0028]      size = 0004 [0x0004]    ----> participants
				 */
				if (V22_JNL_VER <= remote_jnl_ver)
				{
					assert(42 == ((INTPTR_T)&rec->jrec_tcom.num_participants - (INTPTR_T)rec));
					assert(SIZEOF(num_participants_2bytes) == SIZEOF(rec->jrec_tcom.num_participants));
					num_participants_2bytes = rec->jrec_tcom.num_participants;
					rec->jrec_tcom.num_participants = GTM_BYTESWAP_16(num_participants_2bytes);
				} else if (V19_JNL_VER <= remote_jnl_ver)
				{
					ptr = (unsigned char *)rec + 34; /* is offset of update_num in V19 struct_jrec_upd */
					assert(SIZEOF(num_participants_2bytes) == SIZEOF(unsigned short));
					num_participants_2bytes = *(unsigned short *)ptr;
					rec->jrec_tcom.num_participants = GTM_BYTESWAP_16(num_participants_2bytes);
				} else
				{
					assert(V17_JNL_VER <= remote_jnl_ver);
					ptr = (unsigned char *)rec + 40; /* is offset of update_num in V19 struct_jrec_upd */
					assert(SIZEOF(num_participants_4bytes) == SIZEOF(uint4));
					num_participants_4bytes = *(uint4 *)ptr;
					rec->jrec_tcom.num_participants = GTM_BYTESWAP_32(num_participants_4bytes);
				}
				assert(rec->jrec_tcom.num_participants);
				/* token_seq.jnl_seqno & strm_seqno have already been endian converted. */
			}
			/* else if (JRT_NULL == rectype)
			 *	token_seq.jnl_seqno & strm_seqno have already been endian converted.
			 */
			assert(reclen == REC_LEN_FROM_SUFFIX(jb, reclen));
		}
		jb = jb + reclen;
		assert(jb == jstart + reclen);
		jlen -= reclen;
	}
	if ((-1 != status) && (0 != jlen))
	{	/* Incomplete record */
		assert(FALSE);
		status = -1;
	}
	return status;
}

STATICFNDEF void do_flow_control(gtm_uint64_t write_pos)
{
	/* Check for overflow before writing */

	recvpool_ctl_ptr_t	recvpool_ctl;
	upd_proc_local_ptr_t	upd_proc_local;
	gtmrecv_local_ptr_t	gtmrecv_local;
	long			space_used;
	unsigned char		*msg_ptr;				/* needed for REPL_{SEND,RECV}_LOOP */
	int			tosend_len, sent_len, sent_this_iter;	/* needed for REPL_SEND_LOOP */
	int			torecv_len, recvd_len, recvd_this_iter;	/* needed for REPL_RECV_LOOP */
	int			status, poll_dir;			/* needed for REPL_{SEND,RECV}_LOOP */
	gtm_uint64_t		read_pos;
	seq_num			temp_seq_num;
	DCL_THREADGBL_ACCESS;

	SETUP_THREADGBL_ACCESS;
	recvpool_ctl = recvpool.recvpool_ctl;
	upd_proc_local = recvpool.upd_proc_local;
	gtmrecv_local = recvpool.gtmrecv_local;
	space_used = 0;
	if (recvpool_ctl->wrapped)
		space_used = write_pos + recvpool_size - (read_pos = upd_proc_local->read);
	if (!recvpool_ctl->wrapped || space_used > recvpool_size)
		space_used = write_pos - (read_pos = upd_proc_local->read);
	assert(remote_side->endianness_known);	/* only then is remote_side->cross_endian reliable */
	if (space_used >= recvpool_high_watermark && !xoff_sent)
	{	/* Send XOFF message */
		if (!remote_side->cross_endian)
		{
			xoff_msg.type = REPL_XOFF;
			memcpy((uchar_ptr_t)&xoff_msg.msg[0], (uchar_ptr_t)&upd_proc_local->read_jnl_seqno, SIZEOF(seq_num));
			xoff_msg.len = MIN_REPL_MSGLEN;
		} else
		{
			xoff_msg.type = GTM_BYTESWAP_32(REPL_XOFF);
			temp_seq_num = GTM_BYTESWAP_64(upd_proc_local->read_jnl_seqno);
			memcpy((uchar_ptr_t)&xoff_msg.msg[0], (uchar_ptr_t)&temp_seq_num, SIZEOF(seq_num));
			xoff_msg.len = GTM_BYTESWAP_32(MIN_REPL_MSGLEN);
		}
		REPL_SEND_LOOP(gtmrecv_sock_fd, &xoff_msg, MIN_REPL_MSGLEN, REPL_POLL_NOWAIT)
		{
			GTMRECV_POLL_ACTIONS(data_len, buff_unprocessed, buffp);
		}
		CHECK_REPL_SEND_LOOP_ERROR(status, "REPL_XOFF");
		if (gtmrecv_logstats)
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Space used = %ld, High water mark = %ld Low water mark = %ld, "
					"Updproc Read = %ld, Recv Write = %ld, Sent XOFF\n", space_used, recvpool_high_watermark,
					recvpool_low_watermark, read_pos, write_pos);
		repl_log(gtmrecv_log_fp, TRUE, TRUE, "REPL_XOFF sent as receive pool has %ld bytes transaction data yet to be "
				"processed\n", space_used);
		xoff_sent = TRUE;
		xoff_msg_log_cnt = 1;
	} else if (space_used < recvpool_low_watermark && xoff_sent)
	{
		if (!remote_side->cross_endian)
		{
			xon_msg.type = REPL_XON;
			memcpy((uchar_ptr_t)&xon_msg.msg[0], (uchar_ptr_t)&upd_proc_local->read_jnl_seqno, SIZEOF(seq_num));
			xon_msg.len = MIN_REPL_MSGLEN;
		} else
		{
			xon_msg.type = GTM_BYTESWAP_32(REPL_XON);
			temp_seq_num = GTM_BYTESWAP_64(upd_proc_local->read_jnl_seqno);
			memcpy((uchar_ptr_t)&xon_msg.msg[0], (uchar_ptr_t)&temp_seq_num, SIZEOF(seq_num));
			xon_msg.len = GTM_BYTESWAP_32(MIN_REPL_MSGLEN);
		}
		REPL_SEND_LOOP(gtmrecv_sock_fd, &xon_msg, MIN_REPL_MSGLEN, REPL_POLL_NOWAIT)
		{
			GTMRECV_POLL_ACTIONS(data_len, buff_unprocessed, buffp);
		}
		CHECK_REPL_SEND_LOOP_ERROR(status, "REPL_XON");
		if (gtmrecv_logstats)
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Space used now = %ld, High water mark = %ld, "
				 "Low water mark = %ld, Updproc Read = %ld, Recv Write = %ld, Sent XON\n", space_used,
				 recvpool_high_watermark, recvpool_low_watermark, read_pos, write_pos);
		repl_log(gtmrecv_log_fp, TRUE, TRUE, "REPL_XON sent as receive pool has %ld bytes free space to buffer transaction "
				"data\n", recvpool_size - space_used);
		xoff_sent = FALSE;
		xoff_msg_log_cnt = 0;
	}
	return;
}

STATICFNDEF int gtmrecv_est_conn(void)
{
	recvpool_ctl_ptr_t	recvpool_ctl;
	upd_proc_local_ptr_t	upd_proc_local;
	gtmrecv_local_ptr_t	gtmrecv_local;
	boolean_t     		keepalive;
	GTM_SOCKLEN_TYPE	optlen;
	int			status, keepalive_opt, optval, save_errno;
	struct  linger  	disable_linger = {0, 0};
	char			print_msg[1024];
	struct addrinfo		primary_ai;
	struct sockaddr_storage	primary_sas;
	DCL_THREADGBL_ACCESS;

	SETUP_THREADGBL_ACCESS;
	/* Wait for a connection from a Source Server. The Receiver Server is an iterative server. */
	recvpool_ctl = recvpool.recvpool_ctl;
	upd_proc_local = recvpool.upd_proc_local;
	gtmrecv_local = recvpool.gtmrecv_local;
	/* Create a listen socket */
	gtmrecv_comm_init((in_port_t)gtmrecv_local->listen_port);
	primary_ai.ai_addr = (sockaddr_ptr)&primary_sas;
	primary_ai.ai_addrlen = SIZEOF(primary_sas);
	repl_log(gtmrecv_log_fp, TRUE, TRUE, "Waiting for a connection...\n");
	/* Null initialize fields that need to be initialized only after connecting to the primary.
	 * It is ok not to hold a lock on the journal pool while updating jnlpool_ctl fields since this will be the only
	 * process updating those fields.
	 */
	assert(remote_side == &gtmrecv_local->remote_side);
	remote_side->proto_ver = REPL_PROTO_VER_UNINITIALIZED;
	jnlpool->jnlpool_ctl->primary_instname[0] = '\0';
	while (TRUE)
	{
		while (TRUE)
		{
			if (0 < (status = fd_ioready(gtmrecv_listen_sock_fd, REPL_POLLIN, REPL_POLL_WAIT)))
				break;
			if (-1 == status)
			{
				save_errno = errno;
				assert((EAGAIN != save_errno) && (EINTR != save_errno));
				ISSUE_REPLCOMM_ERROR("Error in select on listen socket", save_errno);
			} else if (0 == status)	/* timeout */
				gtmrecv_poll_actions(0, 0, NULL);
		}
		ACCEPT_SOCKET(gtmrecv_listen_sock_fd, primary_ai.ai_addr,
					(GTM_SOCKLEN_TYPE *)&primary_ai.ai_addrlen, gtmrecv_sock_fd);
		if (FD_INVALID == gtmrecv_sock_fd)
		{
			save_errno = errno;
#			ifdef __hpux
			if (ENOBUFS == save_errno)
			{
				gtmrecv_poll_actions(0, 0, NULL);
				continue;
			}
#			endif
			ISSUE_REPLCOMM_ERROR("Error accepting connection from Source Server", save_errno);
		}
		break;
	}
	/* Connection established */
	repl_close(&gtmrecv_listen_sock_fd); /* Close the listener socket */
	repl_connection_reset = FALSE;
	if (-1 == setsockopt(gtmrecv_sock_fd, SOL_SOCKET, SO_LINGER, (const void *)&disable_linger, SIZEOF(disable_linger)))
		ISSUE_REPLCOMM_ERROR("Error with receiver server socket disable linger", errno);
#	ifdef REPL_DISABLE_KEEPALIVE
	keepalive = FALSE;
#	else
	keepalive = TRUE;
#	endif
	if (-1 == setsockopt(gtmrecv_sock_fd, SOL_SOCKET, SO_KEEPALIVE, (const void *)&keepalive,
					SIZEOF(keepalive)))
	{
		ISSUE_REPLCOMM_ERROR("Error with receiver server socket enable keepalive", errno);
	}
	/* Set up the keepalive parameters
	 * TCP_KEEPCNT : overrides tcp_keepalive_probes
	 * TCP_KEEPIDLE: overrides tcp_keepalive_time
	 * TCP_KEEPINTVL: overrides tcp_keepalive_intvl
	 */
#	if defined(KEEPALIVE_PROTO_LEVEL)
	keepalive_opt = KEEPALIVE_PROBES;
	if (-1 == setsockopt(gtmrecv_sock_fd, KEEPALIVE_PROTO_LEVEL, TCP_KEEPCNT, (void*)&keepalive_opt, SIZEOF(keepalive_opt)))
		ISSUE_REPLCOMM_ERROR("Error with receiver server socket setting tcp_keepalive_probes", errno);
	keepalive_opt = KEEPALIVE_TIME;
	if (-1 == setsockopt(gtmrecv_sock_fd, KEEPALIVE_PROTO_LEVEL, TCP_KEEPIDLE, (void*)&keepalive_opt, SIZEOF(keepalive_opt)))
		ISSUE_REPLCOMM_ERROR("Error with receiver server socket setting tcp_keepalive_time", errno);
	keepalive_opt = KEEPALIVE_INTVL;
	if (-1 == setsockopt(gtmrecv_sock_fd, KEEPALIVE_PROTO_LEVEL, TCP_KEEPINTVL, (void*)&keepalive_opt, SIZEOF(keepalive_opt)))
		ISSUE_REPLCOMM_ERROR("Error with receiver server socket setting tcp_keepalive_intvl", errno);
#	endif
	optlen = SIZEOF(optval);
   	if ( -1 == getsockopt(gtmrecv_sock_fd, SOL_SOCKET, SO_KEEPALIVE, &optval, &optlen))
		ISSUE_REPLCOMM_ERROR("Error with receiver server socket checking keepalive enabled or not", errno)
#	if !defined(KEEPALIVE_PROTO_LEVEL)
	repl_log(gtmrecv_log_fp, TRUE, TRUE, "SO_KEEPALIVE is %s\n", (optval ? "ON" : "OFF"));
#	else
	repl_log(gtmrecv_log_fp, TRUE, TRUE, "SO_KEEPALIVE is %s. ", (optval ? "ON" : "OFF"));

	if (-1 == getsockopt(gtmrecv_sock_fd, KEEPALIVE_PROTO_LEVEL, TCP_KEEPCNT, &optval, &optlen))
		ISSUE_REPLCOMM_ERROR("Error with receiver server socket getting tcp_keepalive_probes", errno);
	if (optval)
		repl_log(gtmrecv_log_fp, FALSE, TRUE, "TCP_KEEPCNT is %d, ", optval);

	if (-1 == getsockopt(gtmrecv_sock_fd, KEEPALIVE_PROTO_LEVEL, TCP_KEEPIDLE, &optval, &optlen))
		ISSUE_REPLCOMM_ERROR("Error with receiver server socket getting tcp_keepalive_time", errno);
	if (optval)
       		repl_log(gtmrecv_log_fp, FALSE, TRUE, "TCP_KEEPIDLE is %d, ", optval);

	if (-1 == getsockopt(gtmrecv_sock_fd, KEEPALIVE_PROTO_LEVEL, TCP_KEEPINTVL, &optval, &optlen))
		ISSUE_REPLCOMM_ERROR("Error with receiver server socket getting tcp_keepalive_intvl", errno);
	if (optval)
		repl_log(gtmrecv_log_fp, FALSE, TRUE, "TCP_KEEPINTVL is %d.\n", optval);
#	endif
	if (0 != (status = get_send_sock_buff_size(gtmrecv_sock_fd, &repl_max_send_buffsize)))
		ISSUE_REPLCOMM_ERROR("Error getting socket send buffsize", errno);
	if (0 != (status = get_recv_sock_buff_size(gtmrecv_sock_fd, &repl_max_recv_buffsize)))
		ISSUE_REPLCOMM_ERROR("Error getting socket recv buffsize", errno);
	repl_log(gtmrecv_log_fp, TRUE, TRUE, "Connection established, using TCP send buffer size %d receive buffer size %d\n",
			repl_max_send_buffsize, repl_max_recv_buffsize);
	repl_log_conn_info(gtmrecv_sock_fd, gtmrecv_log_fp, FALSE);
	/* re-determine endianness of other side */
	remote_side->endianness_known = FALSE;
	/* re-determine journal format of other side */
	remote_side->jnl_ver = 0;
	/* re-determine compression level on the replication pipe after every connection establishment */
	repl_zlib_cmp_level = ZLIB_CMPLVL_NONE;
	/* Reset prior connection related state variables (see <C9J02_003091_receiver_server_assert_due_to_lingering_XOFF>) */
	xoff_sent = FALSE;
	xoff_msg_log_cnt = 0;
#	ifdef GTM_TLS
	assert(!repl_tls.enabled);
	/* We either did not create a TLS/SSL aware socket or the SSL object off of the TLS/SSL aware socket should be NULL since
	 * we haven't yet connected to the source server. Assert that.
	 */
	assert((NULL == repl_tls.sock) || (NULL == repl_tls.sock->ssl));
#	endif
	/* Note that even though we are reopening a fresh connection, we should NOT reset the cached information
	 * last_rcvd_histinfo, last_valid_histinfo etc. in this case as we might just resume processing from where
	 * the previous connection left off in which case all the cached information is still valid. If we don't
	 * resume from where we left off, the receiver will anyways error out asking for a rollback to be done so
	 * the cached information never gets used if it is invalid.
	 */
	return (SS_NORMAL);
}

int gtmrecv_alloc_filter_buff(int bufsiz)
{
	unsigned char	*old_filter_buff;

	bufsiz = ROUND_UP2(bufsiz, OS_PAGE_SIZE);
	if ((NO_FILTER != gtmrecv_filter) && (repl_filter_bufsiz < bufsiz))
	{
		REPL_DPRINT3("Expanding filter buff from %d to %d\n", repl_filter_bufsiz, bufsiz);
		old_filter_buff = repl_filter_buff;
		repl_filter_buff = malloc(bufsiz);
		if (NULL != old_filter_buff)
		{
			memcpy(repl_filter_buff, old_filter_buff, repl_filter_bufsiz);
			free(repl_filter_buff);
		}
		repl_filter_bufsiz = bufsiz;
	}
	return (SS_NORMAL);
}

void gtmrecv_free_filter_buff(void)
{
	if (NULL != repl_filter_buff)
	{
		free(repl_filter_buff);
		repl_filter_buff = NULL;
		repl_filter_bufsiz = 0;
	}
}

int gtmrecv_alloc_msgbuff(void)
{
	gtmrecv_max_repl_msglen = MAX_REPL_MSGLEN;
	assert(NULL == gtmrecv_msgp); /* first time initialization. The receiver server doesn't need to re-allocate */
	msgbuff = (unsigned char *)malloc(gtmrecv_max_repl_msglen + OS_PAGE_SIZE);
	gtmrecv_msgp = (repl_msg_ptr_t)ROUND_UP2((unsigned long)msgbuff, OS_PAGE_SIZE);
	gtmrecv_alloc_filter_buff(gtmrecv_max_repl_msglen);
	return (SS_NORMAL);
}

void gtmrecv_free_msgbuff(void)
{
	if (NULL != msgbuff)
	{
		assert(NULL != gtmrecv_msgp);
		free(msgbuff);
		msgbuff = NULL;
		gtmrecv_msgp = NULL;
	}
}

/* This function can be used to only send fixed-size message types across the replication pipe.
 * This in turn uses REPL_SEND* macros but also does error checks and sets the global variables
 *	"repl_connection_reset" or "gtmrecv_wait_for_jnl_seqno" accordingly.
 * It also does the endian conversion of the 'type' and 'len' fields of the repl_msg_t structure being sent.
 *
 *	msg            = Pointer to the message buffer to send
 *	type	       = One of the various message types listed in repl_msg.h
 *	len	       = Length of the message to be sent
 *	msgtypestr     = Message name as a string to display meaningful error messages
 *	optional_seqno = Optional seqno that needs to be printed along with the message name
 */
void	gtmrecv_repl_send(repl_msg_ptr_t msgp, int4 type, int4 len, char *msgtypestr, seq_num optional_seqno)
{
	unsigned char		*msg_ptr;				/* needed for REPL_SEND_LOOP */
	int			tosend_len, sent_len, sent_this_iter;	/* needed for REPL_SEND_LOOP */
	int			status, poll_dir;			/* needed for REPL_{SEND,RECV}_LOOP */
	FILE			*log_fp;
	DCL_THREADGBL_ACCESS;

	SETUP_THREADGBL_ACCESS;
	assert(!jgbl.mur_rollback || !jgbl.mur_options_forward); /* ROLLBACK -FORWARD should not call this function */
	assert(!mur_options.rollback || (NULL == recvpool.gtmrecv_local));
	assert(mur_options.rollback || (NULL != recvpool.gtmrecv_local));
	assert((REPL_MULTISITE_MSG_START > type) || (REPL_PROTO_VER_MULTISITE <= remote_side->proto_ver));
	log_fp = (NULL == gtmrecv_log_fp) ? stdout : gtmrecv_log_fp;
	if (MAX_SEQNO != optional_seqno)
	{
		repl_log(log_fp, TRUE, TRUE, "Sending %s message with seqno "INT8_FMT" "INT8_FMTX"\n", msgtypestr,
			optional_seqno, optional_seqno);
	} else
		repl_log(log_fp, TRUE, TRUE, "Sending %s message\n", msgtypestr);
	/* Assert that if we don't know the endianness of the remote side, we assume it is the same endianness */
	assert(remote_side->endianness_known || !remote_side->cross_endian);
	if (!remote_side->cross_endian)
	{
		msgp->type = type;
		msgp->len = len;
	} else
	{
		msgp->type = GTM_BYTESWAP_32(type);
		msgp->len = GTM_BYTESWAP_32(len);
	}
	REPL_SEND_LOOP(gtmrecv_sock_fd, msgp, len, REPL_POLL_NOWAIT)
	{
		GTMRECV_POLL_ACTIONS(data_len, buff_unprocessed, buffp);
	}
	CHECK_REPL_SEND_LOOP_ERROR(status, msgtypestr);
	assert(SS_NORMAL == status);
}

/* This function is invoked on receipt of a REPL_NEED_INSTINFO message.
 * After doing some checks, it sends back a REPL_INSTINFO message containing the instance information.
 * If any of the checks fail it issues the appropriate error right away.
 * There can be TWO callers. One is the receiver server and another fetchresync rollback.
 *	is_rcvr_srvr is 1 for the former and 0 for the latter.
 */
void	gtmrecv_check_and_send_instinfo(repl_needinst_msg_ptr_t need_instinfo_msg, boolean_t is_rcvr_srvr)
{
	boolean_t		remote_side_is_supplementary, grab_lock_needed, lms_group_different;
	repl_inst_hdr_ptr_t	inst_hdr;
	repl_instinfo_msg_t	instinfo_msg;
	repl_inst_uuid		*strm_start, *strm_top, *strm_info;
	FILE			*log_fp;
	int			reuse_slot, first_usable_slot;
	uint4			creator_pid;
	seq_num			strm_jnl_seqno;
	sgmnt_addrs		*repl_csa;
	DCL_THREADGBL_ACCESS;

	SETUP_THREADGBL_ACCESS;
	assert(!jgbl.mur_rollback || !jgbl.mur_options_forward); /* ROLLBACK -FORWARD should not call this function */
	repl_csa = &FILE_INFO(jnlpool->jnlpool_dummy_reg)->s_addrs;
	grab_lock_needed = is_rcvr_srvr || (jnlpool && (NULL != jnlpool->jnlpool_ctl) && !repl_csa->hold_onto_crit);
	remote_side_is_supplementary = need_instinfo_msg->is_supplementary;
	remote_side->is_supplementary = remote_side_is_supplementary;
	assert(remote_side->endianness_known); /* ensure remote_side->cross_endian is reliable */
	/* If the remote source server is <= V62000 it would have endian converted "need_instinfo_msg" ONLY IF
	 * src_jnl_ver < rcv_jnl_ver. If remote source server is > V62000 it would have endian converted unconditionally.
	 * So we don't need to do the endian conversion in either of those cases. The need_instinfo_msg->proto_ver will let
	 * us distinguish the > V62000 vs <= V62000 case. But if src version is <= V62000, we cannot easily distinguish what
	 * the other side's jnl_ver is so we don't know if the source server would have endian converted or not. We work around
	 * it by assuming that pids are limited to a max of 2**24 and so the most significant byte of the 4-byte pid should be 0.
	 * This trick might not work always in case an OS supports > 2**24 pid number (I don't know of any now) OR if both the
	 * most and least significant bytes of the 4-byte pid are 0, but if it does not, the worst we will see is a) an incorrect
	 * INSNOTJOINED error issued by the receiver server OR b) an incorrect lms_group_info copied over into the instance file
	 * (which will soon elicit some other related error). We can address (a) by trying out endian converting again below to
	 * see if that prevents the error and if so use that instead but we cannot address (b) with this trick.
	 * We can live with since the fix for (b) is to upgrade the receiver side to GTM V62001 or higher.
	 */
	if (remote_side->cross_endian && (REPL_PROTO_VER_XENDIANFIXES > need_instinfo_msg->proto_ver))
	{
		creator_pid = need_instinfo_msg->lms_group_info.creator_pid;
		if (!(creator_pid & 0xff) && (creator_pid & 0xff000000))
			ENDIAN_CONVERT_REPL_INST_UUID(&need_instinfo_msg->lms_group_info);
	}
	assert((is_rcvr_srvr && (NULL != gtmrecv_log_fp)) || (!is_rcvr_srvr && (NULL == gtmrecv_log_fp)));
	assert(!is_rcvr_srvr || !repl_csa->hold_onto_crit);
	assert(is_rcvr_srvr || !jgbl.onlnrlbk || repl_csa->hold_onto_crit);
	log_fp = !is_rcvr_srvr ? stdout : gtmrecv_log_fp;
	repl_log(log_fp, TRUE, TRUE, "Received REPL_NEED_INSTINFO message from primary instance [%s]\n",
		need_instinfo_msg->instname);
	inst_hdr = jnlpool->repl_inst_filehdr;
	assert(NULL != inst_hdr);
	/* The fact that we came here (REPL_NEED_INSTINFO message) implies the source server understands the
	 * supplementary protocol. In that case, make sure -UPDATERESYNC if specified at receiver server startup
	 * had a value as well. If not, issue error.
	 */
	if (is_rcvr_srvr && recvpool.gtmrecv_local->updateresync && (FD_INVALID == recvpool.gtmrecv_local->updresync_instfile_fd))
	{
		rts_error_csa(CSA_ARG(NULL) VARLSTCNT(6) ERR_UPDSYNCINSTFILE, 0, ERR_TEXT, 2,
			  LEN_AND_LIT("Source side is >= V5.5-000 implies -UPDATERESYNC needs a value specified"));
		assert(FALSE);	/* we don't expect the rts_error to return control */
	}
	/* We usually expect the LMS group info to be non-NULL on both primary and secondary. An exception is if
	 * both of them are being brought up for the first time using a GT.M version that supports supplementary instances.
	 * In this case, if the primary was brought up as a root primary, then it should still have the group info filled.
	 * Assert that.
	 */
	assert(inst_hdr->lms_group_info.created_time
		|| need_instinfo_msg->lms_group_info.created_time || !need_instinfo_msg->is_rootprimary);
	assert((is_rcvr_srvr && (NULL != jnlpool->jnlpool_ctl)) || (!is_rcvr_srvr && (NULL == jnlpool->jnlpool_ctl))
			|| jgbl.onlnrlbk || (jgbl.mur_rollback && INST_FREEZE_ON_ERROR_POLICY));
	/* If this instance is supplementary and the journal pool exists (to indicate whether updates are enabled or not
	 * which in turn helps us know whether this is an originating instance or not) do some additional checks.
	 */
	if (is_rcvr_srvr && inst_hdr->is_supplementary)
	{
		assert(NULL != jnlpool->jnlpool_ctl);
		if (!jnlpool->jnlpool_ctl->upd_disabled)
		{	/* this supplementary instance was started with -UPDOK. Issue error if source is also supplementary */
			if (need_instinfo_msg->is_supplementary)
			{
				rts_error_csa(CSA_ARG(NULL) VARLSTCNT(6) ERR_NOSUPPLSUPPL, 4,
						LEN_AND_STR((char *)inst_hdr->inst_info.this_instname),
						LEN_AND_STR((char *)need_instinfo_msg->instname));
				assert(FALSE);	/* we don't expect the rts_error to return control */
			}
		} else
		{	/* this supplementary instance was started with -UPDNOTOK. Issue error if source is not supplementary */
			if (!need_instinfo_msg->is_supplementary)
			{
				rts_error_csa(CSA_ARG(NULL) VARLSTCNT(6) ERR_SUPRCVRNEEDSSUPSRC, 4,
					LEN_AND_STR((char *)inst_hdr->inst_info.this_instname),
					LEN_AND_STR((char *)need_instinfo_msg->instname));
				assert(FALSE);	/* we don't expect the rts_error to return control */
			}
		}
	}
	/* Assert that if the receiver side is a root primary (i.e. has updates enabled), it better be a
	 * supplementary instance and have the LMS group info filled in. Exception is if this a FETCHRESYNC
	 * ROLLBACK run on an instance which was once primary (not necessarily a supplementary one).
	 */
	assert((NULL == jnlpool->jnlpool_ctl) || jnlpool->jnlpool_ctl->upd_disabled
		|| (inst_hdr->is_supplementary && IS_REPL_INST_UUID_NON_NULL(inst_hdr->lms_group_info))
		|| jgbl.onlnrlbk || (jgbl.mur_rollback && INST_FREEZE_ON_ERROR_POLICY));
	/* Check if primary and secondary are in same LMS group. Otherwise issue error. An exception is if the group info has
	 * not yet been filled in after instance file creation. In that case, copy the info from primary and skip the error check.
	 */
	if (IS_REPL_INST_UUID_NON_NULL(inst_hdr->lms_group_info))
	{	/* LMS Group info has been initialized. Compare with that of the source side */
		if (memcmp(&need_instinfo_msg->lms_group_info, &inst_hdr->lms_group_info, SIZEOF(inst_hdr->lms_group_info)))
		{	/* Source and Receiver are part of DIFFERENT LMS Groups. If this instance is supplementary and remote
			 * side is not supplementary then we expect them to be different. Otherwise issue error.
			 * Note that because of an issue with endian-conversion (see big comment at start of this function),
			 * it is possible this lms_group_info would be good (i.e. passed the memcmp) if endian-converted. So
			 * try that once and if it still fails, go ahead with issuing an error.
			 */
			lms_group_different = TRUE;
			if (remote_side->cross_endian && (REPL_PROTO_VER_XENDIANFIXES > need_instinfo_msg->proto_ver))
			{
				ENDIAN_CONVERT_REPL_INST_UUID(&need_instinfo_msg->lms_group_info);
				if (!memcmp(&need_instinfo_msg->lms_group_info,
						&inst_hdr->lms_group_info, SIZEOF(inst_hdr->lms_group_info)))
					lms_group_different = FALSE;	/* stay with converted lms_group_info as it is good */
				else
				{	/* neither works so return to the original */
					ENDIAN_CONVERT_REPL_INST_UUID(&need_instinfo_msg->lms_group_info);
				}
			}
		} else
			lms_group_different = FALSE;
		if (lms_group_different)
		{
			if (!inst_hdr->is_supplementary || remote_side_is_supplementary)
			{
				rts_error_csa(CSA_ARG(NULL) VARLSTCNT(6) ERR_INSNOTJOINED, 4,
						LEN_AND_STR((char *)inst_hdr->inst_info.this_instname),
						LEN_AND_STR((char *)need_instinfo_msg->instname));
				assert(FALSE);	/* we don't expect the rts_error to return control */
			}
		} else
		{	/* Primary and Secondary are part of SAME LMS Group. If this instance is supplementary and remote
			 * side is not supplementary then we expect them to be different. Issue error in that case.
			 */
			if (inst_hdr->is_supplementary && !remote_side_is_supplementary)
			{
				assert(!need_instinfo_msg->is_supplementary); /* else NOSUPPLSUPPL error must have been issued */
				rts_error_csa(CSA_ARG(NULL) VARLSTCNT(6) ERR_INSROLECHANGE, 4,
						LEN_AND_STR((char *)inst_hdr->inst_info.this_instname),
						LEN_AND_STR((char *)need_instinfo_msg->instname));
				assert(FALSE);	/* we don't expect the rts_error to return control */
			}
		}
		if (is_rcvr_srvr && recvpool.gtmrecv_local->updateresync
			&& (FD_INVALID != recvpool.gtmrecv_local->updresync_instfile_fd))
		{	/* Caller is receiver server and -UPDATERESYNC=<INSTFILENAME> was specified. Check if the lms_group_info
			 * of that file matches that of the source side. If not issue error.
			 */
			if (memcmp(&recvpool.gtmrecv_local->updresync_lms_group,
						&need_instinfo_msg->lms_group_info, SIZEOF(inst_hdr->lms_group_info)))
			{
				RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(6) ERR_UPDSYNCINSTFILE, 0, ERR_TEXT, 2,
					LEN_AND_LIT("Specified input instance file does not have same "
					"LMS Group information as source server instance"));
			}
		}
	} else if (IS_REPL_INST_UUID_NON_NULL(need_instinfo_msg->lms_group_info))
	{	/* LMS Group info is NULL in instance file header. Initialize it in journal pool from the value
		 * on the primary side AND flush the changes to disk. Get lock before manipulating it.
		 * If caller is rollback, no other process can be touching the instance file until we are done so
		 * no need of the lock in that case.
		 */
		if (grab_lock_needed)
		{
			grab_lock(jnlpool->jnlpool_dummy_reg, TRUE, GRAB_LOCK_ONLY);
			if (is_rcvr_srvr)
				GTMRECV_ONLN_RLBK_CLNUP_IF_NEEDED;
		}
		inst_hdr->lms_group_info = need_instinfo_msg->lms_group_info;
		repl_inst_flush_filehdr();
		if (grab_lock_needed)
			rel_lock(jnlpool->jnlpool_dummy_reg);
	}
	/* If this instance is supplementary and remote side is not, then find out which stream # the non-supplementary source
	 * corresponds to. Issue error if the source LMS group is unknown in the instance file. If this is non-supplementary,
	 * the stream index is 0.
	 */
	if (inst_hdr->is_supplementary && !remote_side_is_supplementary)
	{
		assert(ARRAYSIZE(inst_hdr->strm_group_info) == (MAX_SUPPL_STRMS - 1));
		assert(SIZEOF(repl_inst_uuid) == SIZEOF(inst_hdr->strm_group_info[0]));
		strm_start = &inst_hdr->strm_group_info[0];
		strm_top = strm_start + ARRAYSIZE(inst_hdr->strm_group_info);
		/* Do this check under a lock as another receiver server started with -UPDATERESYNC=<INSTFILENAME> (not supported
		 * now but will be in the future) could be changing the "strm_group_info" array concurrently.
		 * If a matching stream is found, update gtmrecv_local->strm_index to reflect this while still holding lock.
		 * This field is checked by the -UPDATERESYNC to see if a receiver has a given stream # actively in use.
		 */
		if (grab_lock_needed)
		{
			grab_lock(jnlpool->jnlpool_dummy_reg, TRUE, GRAB_LOCK_ONLY);
			if (is_rcvr_srvr)
				GTMRECV_ONLN_RLBK_CLNUP_IF_NEEDED;
		}
		reuse_slot = 0;
		if (is_rcvr_srvr && recvpool.gtmrecv_local->updateresync && gtmrecv_options.reuse_specified)
		{	/* If -REUSE was specified, check if instance name specified matches any existing slot.
			 * If slot has already been found, don't search any more.
			 */
			for (strm_info = strm_start; strm_info < strm_top; strm_info++)
			{
				if (IS_REPL_INST_UUID_NULL(*strm_info))
					continue;
				if (!STRCMP(gtmrecv_options.reuse_instname, strm_info->this_instname))
				{
					reuse_slot = (strm_info - strm_start) + 1;
					break;
				}
			}
			if (strm_info == strm_top)
			{	/* -REUSE specified an instance name that is not present in any of the 15 strm_group slots */
				rel_lock(jnlpool->jnlpool_dummy_reg);
				RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(6) ERR_REUSEINSTNAME, 0, ERR_TEXT, 2,
					LEN_AND_LIT("Instance name in REUSE does not match any of 15 slots in instance file"));
			}
			assert(reuse_slot);
		}
		first_usable_slot = 0;
		strm_index = 0;
		for (strm_info = strm_start; strm_info < strm_top; strm_info++)
		{
			if (IS_REPL_INST_UUID_NULL(*strm_info))
			{
				if (!first_usable_slot)
					first_usable_slot = (strm_info - strm_start) + 1;
				continue;
			}
			if (!memcmp(&need_instinfo_msg->lms_group_info, strm_info, SIZEOF(repl_inst_uuid)))
			{	/* Found the stream corresponding to the source side */
				strm_index = (strm_info - strm_start) + 1;
				break;
			}
		}
		if (strm_info == strm_top)
		{	/* Non-supplementary source is unknown to this supplementary instance */
			if (reuse_slot)	/* -REUSE specified and did locate a reusable slot. Use it. */
			{
				assert(is_rcvr_srvr);
				strm_index = reuse_slot;
				/* having reused the slot as specified, the qualifier has no further use & may be disruptive */
				gtmrecv_options.reuse_specified = FALSE;
			} else if (gtmrecv_options.resume_specified)
			{
				assert(is_rcvr_srvr);
				strm_index = gtmrecv_options.resume_strm_num;
			} else if (is_rcvr_srvr && recvpool.gtmrecv_local->updateresync)
			{
				if (first_usable_slot)
					strm_index = first_usable_slot;
				else
				{
					if (grab_lock_needed)
						rel_lock(jnlpool->jnlpool_dummy_reg);
					RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(6) ERR_UPDSYNCINSTFILE, 0, ERR_TEXT, 2,
						LEN_AND_LIT("No empty slot found. Specify REUSE to choose one for reuse"));
				}
			} else
			{
				if (grab_lock_needed)
					rel_lock(jnlpool->jnlpool_dummy_reg);
				rts_error_csa(CSA_ARG(NULL) VARLSTCNT(6) ERR_INSUNKNOWN, 4,
						LEN_AND_STR((char *)inst_hdr->inst_info.this_instname),
						LEN_AND_STR((char *)need_instinfo_msg->instname));
				assert(FALSE);	/* we don't expect the rts_error to return control */
			}
			/* Since we did not find the stream in the existing instance file but did find a slot, fill that slot
			 * while we have the lock on the instance file. This way another -updateresync startup of a receiver
			 * server (when we have multiple receiver server support) will see this slot taken when it gets the lock.
			 */
			assert(is_rcvr_srvr && recvpool.gtmrecv_local->updateresync
				&& (FD_INVALID != recvpool.gtmrecv_local->updresync_instfile_fd));
			assert(!memcmp(&recvpool.gtmrecv_local->updresync_lms_group,
						&need_instinfo_msg->lms_group_info, SIZEOF(inst_hdr->lms_group_info)));
			assert(0 < strm_index);
			strm_info = &strm_start[strm_index - 1];
			assert(strm_info < strm_top);
			assert(IS_REPL_INST_UUID_NON_NULL(need_instinfo_msg->lms_group_info));
			*strm_info = need_instinfo_msg->lms_group_info;
			repl_inst_flush_filehdr();
		} else if ((gtmrecv_options.resume_specified) && (gtmrecv_options.resume_strm_num != strm_index))
		{	/* If -RESUME was specified, then the slot it matched must be same as slot found without its use */
			assert(is_rcvr_srvr);
			rel_lock(jnlpool->jnlpool_dummy_reg);
			RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(6) ERR_RESUMESTRMNUM, 0, ERR_TEXT, 2, LEN_AND_LIT("Source side LMS "
				"group is found in instance file but RESUME specifies different stream number"));
		} else if (reuse_slot && (reuse_slot != strm_index))
		{	/* If -REUSE was specified, then the slot it matched must be same as slot found without its use */
			assert(is_rcvr_srvr);
			rel_lock(jnlpool->jnlpool_dummy_reg);
			RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(6) ERR_REUSEINSTNAME, 0, ERR_TEXT, 2, LEN_AND_LIT("Source side LMS "
				"group is found in instance file but REUSE specifies different instance name"));
		}
		assert(INVALID_SUPPL_STRM != strm_index);
		assert(0 < strm_index);
		assert(MAX_SUPPL_STRMS > strm_index);
		/* Maintain stream slot # in shared memory as well so another -UPDATERESYNC=<INSTFILENAME>
		 * can see which stream# is actively in use by this receiver server. */
		if (is_rcvr_srvr)
		{
			if (recvpool.gtmrecv_local->strm_index && (strm_index != recvpool.gtmrecv_local->strm_index))
			{	/* This receiver server has already connected to a source server with a different stream #.
				 * Since mixing of multiple stream journal records in the same receive pool confuses the
				 * update process, issue error. Note: This limitation "might" be removed in the future.
				 */
				rel_lock(jnlpool->jnlpool_dummy_reg);
				RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(4) ERR_RCVRMANYSTRMS, 2,
					strm_index, recvpool.gtmrecv_local->strm_index);
			}
			recvpool.gtmrecv_local->strm_index = strm_index;
			repl_log(gtmrecv_log_fp, TRUE, TRUE,
					"Determined non-supplementary source Stream # = %d\n", strm_index);
			assert(IS_REPL_INST_UUID_NON_NULL(need_instinfo_msg->lms_group_info));
			recvpool.gtmrecv_local->remote_lms_group = need_instinfo_msg->lms_group_info;
			rel_lock(jnlpool->jnlpool_dummy_reg);
		} else
			repl_log(stdout, TRUE, TRUE, "Determined non-supplementary source Stream # = %d\n", strm_index);
		/* Compute non-zero strm_jnl_seqno to send across in the REPL_INSTINFO message.
		 * For receiver server
		 *	If updateresync startup and this is the first connection to a source server
		 *		If -RESUME is not specified :
		 *			--> gtmrecv_local->updresync_jnl_seqno
		 *		If -RESUME is specified :
		 *			--> jnlpool_ctl->strm_seqno[gtmrecv_options.resume_strm_num]
		 *	If updateresync startup and this is not the first connection to a source server
		 *		--> recvpool_ctl->jnl_seqno
		 *	If normal startup and this is the first connection to a source server
		 *		--> inst_hdr->strm_seqno[strm_index] where strm_index > 0
		 *	If this is not the first connection to a source server
		 *		--> recvpool_ctl->jnl_seqno
		 *	In case of instance crash
		 *		--> No need to worry about this case as we would have otherwise issued a
		 *		--> REPLREQROLLBACK error when the source server for this instance started up after the crash.
		 * For rollback
		 *	If normal startup after a clean shutdown of the instance file
		 *		--> inst_hdr->strm_seqno[strm_index] where strm_index > 0
		 *	In case of crash shutdown of instance file
		 *		--> Take max of cs_data->strm_seqno[strm_index] across all databases and use this (strm_index > 0)
		 */
		if (is_rcvr_srvr)
		{
			if (recvpool.upd_proc_local->read_jnl_seqno)
				strm_jnl_seqno = recvpool.recvpool_ctl->jnl_seqno;
			else
			{
				if (recvpool.gtmrecv_local->updateresync)
				{
					assert(FD_INVALID != recvpool.gtmrecv_local->updresync_instfile_fd);
					if (!gtmrecv_options.resume_specified)
						strm_jnl_seqno = recvpool.gtmrecv_local->updresync_jnl_seqno;
					else
					{
						strm_jnl_seqno = jnlpool->jnlpool_ctl->strm_seqno[gtmrecv_options.resume_strm_num];
						/* It is possible for the strm_seqno to be 0. This implies the stream has
						 * had no updates. In that case, ideally this value should have been 1.
						 * But because we want to differentiate a stream that has had updates from
						 * stream numbers where there is no interest, we follow this convention.
						 * Therefore, in this case, reset the 0 back to 1 so we never send a zero
						 * seqno to the other side.
						 */
						if (!strm_jnl_seqno)
							strm_jnl_seqno = 1;
					}
				} else
					strm_jnl_seqno = jnlpool->jnlpool_ctl->strm_seqno[strm_index];
				assert(0 == GET_STRM_INDEX(strm_jnl_seqno));
			}
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Sending Stream Seqno = "INT8_FMT" "INT8_FMTX"\n",
										strm_jnl_seqno, strm_jnl_seqno);
		} else
		{
			if (jnlpool->repl_inst_filehdr->crash)
				strm_jnl_seqno = mur_get_max_strm_reg_seqno(strm_index);
			else
			{
				assert((NULL == jnlpool->jnlpool_ctl) || jgbl.onlnrlbk);
				assert((NULL == jnlpool->jnlpool_ctl)
					|| (jnlpool->jnlpool_ctl->strm_seqno[strm_index] == inst_hdr->strm_seqno[strm_index]));
				strm_jnl_seqno = jnlpool->repl_inst_filehdr->strm_seqno[strm_index];
			}
			repl_log(stdout, TRUE, TRUE, "Sending Stream Seqno = "INT8_FMT" "INT8_FMTX"\n",
										strm_jnl_seqno, strm_jnl_seqno);
		}
	} else
	{
		strm_jnl_seqno = 0;	/* actually no need to initialize this since source server will not look at this
					 * field in this case but still be safe */
		if (inst_hdr->is_supplementary)
		{	/* In case caller is receiver server, "strm_index" would have been already set to 0 in jnlpool_init.c.
			 * But in case caller is rollback, "strm_index" would still be set to -1. In this case, set it to 0.
			 */
			assert(!is_rcvr_srvr || (0 == strm_index));
			strm_index = 0;
		}
		DEBUG_ONLY(
			if (is_rcvr_srvr)
				NULL_INITIALIZE_REPL_INST_UUID(recvpool.gtmrecv_local->remote_lms_group);
		)
	}
	assert(!is_rcvr_srvr || (INVALID_SUPPL_STRM == recvpool.gtmrecv_local->strm_index)
		|| ((0 <= recvpool.gtmrecv_local->strm_index) && (MAX_SUPPL_STRMS > recvpool.gtmrecv_local->strm_index)));
	/* Initialize the remote side protocol version from "proto_ver" field of this msg */
	assert(REPL_PROTO_VER_SUPPLEMENTARY <= need_instinfo_msg->proto_ver);
	remote_side->proto_ver = need_instinfo_msg->proto_ver;
	/*************** Send REPL_INSTINFO message ***************/
	memset(&instinfo_msg, 0, SIZEOF(instinfo_msg));
	memcpy(instinfo_msg.instname, inst_hdr->inst_info.this_instname, MAX_INSTNAME_LEN - 1);
	if (grab_lock_needed)
	{
		grab_lock(jnlpool->jnlpool_dummy_reg, TRUE, GRAB_LOCK_ONLY);
		if (is_rcvr_srvr)
			GTMRECV_ONLN_RLBK_CLNUP_IF_NEEDED;
	}
	instinfo_msg.was_rootprimary = (unsigned char)repl_inst_was_rootprimary();
	if (grab_lock_needed)
		rel_lock(jnlpool->jnlpool_dummy_reg);
	if (!is_rcvr_srvr)
		murgbl.was_rootprimary = instinfo_msg.was_rootprimary;
	instinfo_msg.strm_jnl_seqno = strm_jnl_seqno;
	/* strm_jnl_seqno is not expected to be zero unless this is a non-supplementary instance (like A->B) in which case
	 * strm_seqno is not maintained OR the remote side is supplementary (like P->Q) in which case the two instances do
	 * not communicate in-terms of strm_seqno (once the handshake is established)
	 */
	assert(strm_jnl_seqno || !inst_hdr->is_supplementary || remote_side_is_supplementary);
	instinfo_msg.lms_group_info = inst_hdr->lms_group_info;
	assert(remote_side->endianness_known);	/* only then is remote_side->cross_endian reliable */
	if (remote_side->cross_endian)
	{
		ENDIAN_CONVERT_REPL_INST_UUID(&instinfo_msg.lms_group_info);
		instinfo_msg.strm_jnl_seqno = GTM_BYTESWAP_64(instinfo_msg.strm_jnl_seqno);
	}
	gtmrecv_repl_send((repl_msg_ptr_t)&instinfo_msg, REPL_INSTINFO, SIZEOF(repl_instinfo_msg_t), "REPL_INSTINFO", MAX_SEQNO);
	if (repl_connection_reset || (is_rcvr_srvr && gtmrecv_wait_for_jnl_seqno))
		return;
	/* Do not allow an instance which was formerly a root primary or which still
	 * has a non-zero value of "zqgblmod_seqno" to start up as a tertiary. The only exception is
	 * if this is P (supplementary instance) receiving from a non-supplementary instance.
	 * In that case, P can never be a root primary of the non-supplementary group and therefore
	 * cannot be affected by lost transactions being applied from the non-supplementary group.
	 */
	if ((!inst_hdr->is_supplementary || remote_side_is_supplementary)
		&& !need_instinfo_msg->is_rootprimary
		&& (instinfo_msg.was_rootprimary
			|| (is_rcvr_srvr && jnlpool->jnlpool_ctl->max_zqgblmod_seqno)))
	{
		if (is_rcvr_srvr)
		{
			gtm_putmsg_csa(CSA_ARG(NULL) VARLSTCNT(4) ERR_PRIMARYNOTROOT, 2,
					LEN_AND_STR((char *) need_instinfo_msg->instname));
			gtmrecv_autoshutdown();	/* should not return */
		} else
			rts_error_csa(CSA_ARG(NULL) VARLSTCNT(4) ERR_PRIMARYNOTROOT, 2,
					LEN_AND_STR((char *) need_instinfo_msg->instname));
		assert(FALSE);
	}
	if (is_rcvr_srvr)
		memcpy(jnlpool->jnlpool_ctl->primary_instname, need_instinfo_msg->instname, MAX_INSTNAME_LEN - 1);
}

STATICFNDEF int	gtmrecv_start_onln_rlbk(void)
{
	char			command[ONLN_RLBK_CMD_MAXLEN + 1], *errptr;
	int			status, save_errno, cmdlen;
	gtmrecv_local_ptr_t	gtmrecv_local;

	assert(!have_crit(CRIT_HAVE_ANY_REG));
	gtmrecv_local = recvpool.gtmrecv_local;
	MEMCPY_LIT(command, MUPIP_DIST_STR);
	cmdlen = STR_LIT_LEN(MUPIP_DIST_STR);
	MEMCPY_LIT(&command[cmdlen], ONLN_RLBK_CMD);
	cmdlen += STR_LIT_LEN(ONLN_RLBK_CMD);
	if (gtmrecv_options.autorollback_verbose)
	{
		MEMCPY_LIT(&command[cmdlen], ONLN_RLBK_VERBOSE);
		cmdlen += STR_LIT_LEN(ONLN_RLBK_VERBOSE);
	}
	MEMCPY_LIT(&command[cmdlen], ONLN_RLBK_QUALIFIERS);
	cmdlen += STR_LIT_LEN(ONLN_RLBK_QUALIFIERS);
	assert(0 < gtmrecv_local->listen_port);
	SNPRINTF(&command[cmdlen], ONLN_RLBK_CMD_MAXLEN - cmdlen, "%d", gtmrecv_local->listen_port); /* will add '\0' at the end */
	repl_log(gtmrecv_log_fp, TRUE, TRUE, "Executing %s\n", command);
	status = SYSTEM(((char *)command));
	if (0 != status)
	{
		if (-1 == status)
		{
			save_errno = errno;
			errptr = (char *)STRERROR(save_errno);
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "SYSTEM command failed: %s\n", errptr);
		} else
		{	/* ONLINE ROLLBACK returned a non-zero status */
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "ONLINE FETCHRESYNC ROLLBACK exited with code - %d\n", status);
		}
		repl_log(gtmrecv_log_fp, TRUE, TRUE, "Could not complete ONLINE FETCHRESYNC ROLLBACK due to the above errors\n");
	} else
	{
		repl_log(gtmrecv_log_fp, TRUE, TRUE, "ONLINE FETCHRESYNC ROLLBACK completed successfully\n");
	}
	return status;
}

/* We are here because a grab_lock or grab_crit saw a concurrent online rollback and we need to break the connection and
 * re-establish with the new sequence number (the rolled back one). Wait for the update process to let us know the new sequence
 * number from where we should start requesting transactions from.
 * Ideally this function should have been a part of onln_rlbk.c. However, the function calls into gtmrecv_poll_actions which
 * is strictly in the libmupip archive. Moving the function into onln_rlbk.c means it will try to pull in gtmrecv_poll_actions
 * and all the modules that it calls which is something that we don't desire. So, keep this function isolated from onln_rlbk.c
 */
void	gtmrecv_onln_rlbk_clnup(void)
{
	boolean_t		connection_already_reset;

	assert(NULL != gtmrecv_log_fp);
	repl_log(gtmrecv_log_fp, TRUE, TRUE, "---> ONLINE ROLLBACK. Current Jnlpool Seqno : "INT8_FMT"\n",
			jnlpool->jnlpool_ctl->jnl_seqno);
	repl_log(gtmrecv_log_fp, TRUE, TRUE, "Waiting for update process to set recvpool_ctl->onln_rlbk_flag\n");
	connection_already_reset = repl_connection_reset;
	assert(!gtmrecv_wait_for_jnl_seqno);
	while (TRUE)
	{
		SHORT_SLEEP(GTMRECV_WAIT_FOR_UPD_PROGRESS);
		gtmrecv_poll_actions(data_len, buff_unprocessed, buffp);
		/* If connection was already closed before we came here (possible if receiver server closed the connection in
		 * response to a REPL_ROLLBACK_FIRST message) after spawning off an online rollback, then we should NOT check for
		 * repl_connection_reset. This way, we avoid breaking prematurely from this loop without waiting for the update
		 * process to acknowledge the online rollback. In this case, wait for gtmrecv_wait_for_jnl_seqno to be set to TRUE
		 * by gtmrecv_poll_actions.
		 */
		if ((!connection_already_reset && repl_connection_reset) || gtmrecv_wait_for_jnl_seqno)
			break;
	}
	return;
}

/* This function is invoked on receipt of a REPL_NEED_HISTINFO message.
 * This in turn sends a REPL_HISTINFO message containing the history information.
 */
void gtmrecv_send_histinfo(repl_histinfo *cur_histinfo)
{
	repl_histinfo1_msg_t	histinfo1_msg;
	repl_histinfo2_msg_t	histinfo2_msg;
	repl_histinfo_msg_t	histinfo_msg;
	FILE			*log_fp;
	char			remote_proto_ver;
	DCL_THREADGBL_ACCESS;

	SETUP_THREADGBL_ACCESS;
	/* If sending history from a supplementary to a non-supplementary version, assert that the history record
	 * particularly the "strm_seqno" is 0 as a non-zero value is not understood by a non-supplementary instance.
	 */
	assert((NULL == this_side) || (this_side->is_supplementary == jnlpool->repl_inst_filehdr->is_supplementary));
	assert(!jnlpool->repl_inst_filehdr->is_supplementary || remote_side->is_supplementary || !cur_histinfo->strm_seqno);
	remote_proto_ver = remote_side->proto_ver;
	assert(REPL_PROTO_VER_MULTISITE <= remote_proto_ver);
	assert(remote_side->endianness_known);	/* only then is remote_side->cross_endian reliable */
	if (REPL_PROTO_VER_SUPPLEMENTARY > remote_proto_ver)
	{	/* Remote side does not support supplementary protocol. Send OLDER histinfo messages */
		/*************** Send REPL_OLD_TRIPLEINFO1 message ***************/
		histinfo1_msg.start_seqno = !remote_side->cross_endian ?
			cur_histinfo->start_seqno : GTM_BYTESWAP_64(cur_histinfo->start_seqno);
		memcpy(histinfo1_msg.instname, cur_histinfo->root_primary_instname, MAX_INSTNAME_LEN - 1);
		histinfo1_msg.instname[MAX_INSTNAME_LEN - 1] = '\0';
		gtmrecv_repl_send((repl_msg_ptr_t)&histinfo1_msg, REPL_OLD_TRIPLEINFO1, MIN_REPL_MSGLEN,
					"REPL_OLD_TRIPLEINFO1", cur_histinfo->start_seqno);
		if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
			return;
		/*************** Send REPL_OLD_TRIPLEINFO2 message ***************/
		if (!remote_side->cross_endian)
		{
			histinfo2_msg.start_seqno = cur_histinfo->start_seqno;
			histinfo2_msg.cycle = cur_histinfo->root_primary_cycle;
			histinfo2_msg.histinfo_num = cur_histinfo->histinfo_num;
		} else
		{
			histinfo2_msg.start_seqno = GTM_BYTESWAP_64(cur_histinfo->start_seqno);
			histinfo2_msg.cycle = GTM_BYTESWAP_32(cur_histinfo->root_primary_cycle);
			histinfo2_msg.histinfo_num = GTM_BYTESWAP_32(cur_histinfo->histinfo_num);
		}
		gtmrecv_repl_send((repl_msg_ptr_t)&histinfo2_msg, REPL_OLD_TRIPLEINFO2, MIN_REPL_MSGLEN,
					"REPL_OLD_TRIPLEINFO2", cur_histinfo->start_seqno);
		if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
			return;
	} else
	{	/* Remote side does support supplementary protocol. Send NEWER histinfo message. */
		histinfo_msg.history = *cur_histinfo;
		histinfo_msg.history.root_primary_instname[MAX_INSTNAME_LEN - 1] = '\0'; /* just in case */
		if (remote_side->cross_endian)
			ENDIAN_CONVERT_REPL_HISTINFO(&histinfo_msg.history);
		gtmrecv_repl_send((repl_msg_ptr_t)&histinfo_msg, REPL_HISTINFO, SIZEOF(repl_histinfo_msg_t),
					"REPL_HISTINFO", cur_histinfo->start_seqno);
	}
	log_fp = (NULL == gtmrecv_log_fp) ? stdout : gtmrecv_log_fp;
	repl_dump_histinfo(log_fp, TRUE, FALSE, "History sent", cur_histinfo);
}

STATICFNDEF void prepare_recvpool_for_write(gtm_uint64_t datalen, gtm_uint64_t pre_filter_write_len)
{
	recvpool_ctl_ptr_t	recvpool_ctl;

	recvpool_ctl = recvpool.recvpool_ctl;
	if (datalen > recvpool_size)
	{	/* Too large a transaction to be accommodated in the Receive Pool */
		RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(7) ERR_REPLTRANS2BIG, 5, &recvpool_ctl->jnl_seqno,
			&datalen, &pre_filter_write_len, LEN_AND_LIT("Receive"));
	}
	if (write_loc > (recvpool_size - datalen))
	{
		REPL_DEBUG_ONLY(
			if (recvpool_ctl->wrapped)
				REPL_DPRINT1("Update Process too slow. Waiting for it to free up space and wrap\n");
		)
		while (recvpool_ctl->wrapped)
		{	/* Wait till the updproc wraps */
			SHORT_SLEEP(GTMRECV_WAIT_FOR_UPD_PROGRESS);
			GTMRECV_POLL_ACTIONS(data_len, buff_unprocessed, buffp);
		}
		assert(recvpool_ctl->wrapped == FALSE);
		recvpool_ctl->write_wrap = write_wrap = write_loc;
		/* The update process reads (a) "recvpool_ctl->write" first. If "write" is not equal to
		 * "upd_proc_local->read", it then reads (b) "recvpool_ctl->write_wrap" and assumes that
		 * "write_wrap" holds a non-stale value. This is in turn used to compare "temp_read" and
		 * "write_wrap" to determine how much of unprocessed data there is in the receive pool. If
		 * it so happens that the receiver server sets "write_wrap" in the above line to a value
		 * that is lesser than its previous value (possible if in the previous wrap of the pool,
		 * transactions used more portions of the pool than in the current wrap), it is important
		 * that the update process sees the updated value of "write_wrap" as long as it sees the
		 * corresponding update to "write". This is because it will otherwise end up processing
		 * the tail section of the receive pool (starting from the uptodate value of "write" to the
		 * stale value of "write_wrap") that does not contain valid journal data. For this read order
		 * dependency to hold good, the receiver server needs to do a write memory barrier
		 * after updating "write_wrap" but before updating "write".  The update process
		 * will do a read memory barrier after reading "wrapped" but before reading "write".
		 */
		SHM_WRITE_MEMORY_BARRIER;
		/* The update process reads (a) "recvpool_ctl->wrapped" first and then reads (b)
		 * "recvpool_ctl->write". If "wrapped" is TRUE, it assumes that "write" will never hold a stale
		 * value that reflects a corresponding previous state of "wrapped" (i.e. "write" will point to
		 * the beginning of the receive pool, either 0 or a small non-zero value instead of pointing
		 * to the end of the receive pool). For this to hold good, the receiver server needs to do
		 * a write memory barrier after updating "write" but before updating "wrapped".  The update
		 * process will do a read memory barrier after reading "wrapped" but before reading "write".
		 */
		recvpool_ctl->write = write_loc = 0;
		SHM_WRITE_MEMORY_BARRIER;
		recvpool_ctl->wrapped = TRUE;
	}
	DO_FLOW_CONTROL(write_loc);
}

STATICFNDEF void copy_to_recvpool(uchar_ptr_t databuff, gtm_uint64_t datalen)
{
	gtm_uint64_t		upd_read;
	gtm_uint64_t		future_write;
	upd_proc_local_ptr_t	upd_proc_local;
	recvpool_ctl_ptr_t	recvpool_ctl;

	recvpool_ctl = recvpool.recvpool_ctl;
	upd_proc_local = recvpool.upd_proc_local;
	future_write = write_loc + datalen;
	assertpro(future_write >= write_loc); /* prepare_recvpool_for_write should have handled this */
	upd_read = upd_proc_local->read;
	REPL_DEBUG_ONLY(
		if (recvpool_ctl->wrapped && (upd_read <= future_write))
		{
			REPL_DPRINT1("Update Process too slow. Waiting for it to free up space\n");
		}
	)
	while (recvpool_ctl->wrapped && (upd_read <= future_write))
	{	/* Write will cause overflow. Wait till there is more space available */
		SHORT_SLEEP(GTMRECV_WAIT_FOR_UPD_PROGRESS);
		GTMRECV_POLL_ACTIONS(data_len, buff_unprocessed, buffp);
		upd_read = upd_proc_local->read;
	}
	memcpy(recvpool.recvdata_base + write_loc, databuff, datalen);
	write_loc = future_write;
	if (write_loc > write_wrap)
		write_wrap = write_loc;
}

STATICFNDEF void wait_for_updproc_to_clear_backlog(void)
{
	upd_proc_local_ptr_t	upd_proc_local;
	recvpool_ctl_ptr_t	recvpool_ctl;

	recvpool_ctl = recvpool.recvpool_ctl;
	upd_proc_local = recvpool.upd_proc_local;
	while (upd_proc_local->read != recvpool_ctl->write)
	{
		SHORT_SLEEP(GTMRECV_WAIT_FOR_UPD_PROGRESS);
		GTMRECV_POLL_ACTIONS(data_len, buff_unprocessed, buffp);
	}
}

STATICFNDEF void process_tr_buff(int msg_type)
{
	recvpool_ctl_ptr_t		recvpool_ctl;
	seq_num				log_seqno, recv_jnl_seqno, upd_seqno, diff_seqno;
	uint4				in_size, out_size, upd_read, max_strm_histinfo;
	boolean_t			filter_pass = FALSE, is_new_histrec, is_repl_cmpc;
	uchar_ptr_t			save_buffp, save_filter_buff, in_buff;
	int				idx, status, num_strm_histinfo;
	qw_num				msg_total;
	repl_old_triple_jnl_t		old_triple_content;
	uLongf				destlen;
	int				cmpret, cur_data_len, rc;
	repl_msg_ptr_t			msgp, msgp_top;
	int4				histinfo_strm_num;
	uint4				write_len, pre_filter_write_len, pre_filter_write;
	boolean_t			uncmpfail;
	repl_histinfo			*cur_histinfo, *pool_histinfo, tmp_histinfo;
	repl_histrec_jnl_t		*pool_histrec, tmp_histjrec, rcvd_strm_histjrec[MAX_SUPPL_STRMS];
	static boolean_t		first_histrec = TRUE;
	DCL_THREADGBL_ACCESS;

	SETUP_THREADGBL_ACCESS;
	recvpool_ctl = recvpool.recvpool_ctl;
	is_repl_cmpc  = ((REPL_TR_CMP_JNL_RECS == msg_type) || (REPL_TR_CMP_JNL_RECS2 == msg_type));
	is_new_histrec = ((REPL_OLD_TRIPLE == msg_type) || (REPL_HISTREC == msg_type));
	assert(!is_repl_cmpc || !is_new_histrec);	/* HISTINFO records should not be compressed in the pipe */
	if (is_repl_cmpc)
	{
		assert(gtmrecv_max_repl_uncmpmsglen);
		destlen = gtmrecv_max_repl_uncmpmsglen;
		if (ZLIB_CMPLVL_NONE == gtm_zlib_cmp_level)
		{	/* Receiver does not have compression enabled in the first place but yet source server has sent
			 * compressed records. Stop source server from sending compressed records.
			 */
			uncmpfail = TRUE;
		} else
		{
			ZLIB_UNCOMPRESS(gtmrecv_uncmpmsgp, destlen, gtmrecv_cmpmsgp, gtmrecv_repl_cmpmsglen, cmpret);
			GTM_WHITE_BOX_TEST(WBTEST_REPL_TR_UNCMP_ERROR, cmpret, Z_DATA_ERROR);
			recv_jnl_seqno = recvpool_ctl->jnl_seqno;
			switch(cmpret)
			{
				case Z_MEM_ERROR:
					assert(FALSE);
					repl_log(gtmrecv_log_fp, TRUE, TRUE, GTM_ZLIB_Z_MEM_ERROR_STR
						GTM_ZLIB_UNCMP_ERR_SEQNO_STR, recv_jnl_seqno, recv_jnl_seqno);
					break;
				case Z_BUF_ERROR:
					assert(FALSE);
					repl_log(gtmrecv_log_fp, TRUE, TRUE, GTM_ZLIB_Z_BUF_ERROR_STR
						GTM_ZLIB_UNCMP_ERR_SEQNO_STR, recv_jnl_seqno, recv_jnl_seqno);
					break;
				case Z_DATA_ERROR:
					assert(gtm_white_box_test_case_enabled
						&& (WBTEST_REPL_TR_UNCMP_ERROR == gtm_white_box_test_case_number));
					repl_log(gtmrecv_log_fp, TRUE, TRUE, GTM_ZLIB_Z_DATA_ERROR_STR
						GTM_ZLIB_UNCMP_ERR_SEQNO_STR, recv_jnl_seqno, recv_jnl_seqno);
					break;
			}
			uncmpfail = (Z_OK != cmpret);
			if (!uncmpfail)
			{
				GTM_WHITE_BOX_TEST(WBTEST_REPL_TR_UNCMP_ERROR, destlen, gtmrecv_repl_uncmpmsglen - 1);
				if (destlen != gtmrecv_repl_uncmpmsglen)
				{	/* decompression did not yield precompressed data length */
					assert(gtm_white_box_test_case_enabled
						&& (WBTEST_REPL_TR_UNCMP_ERROR == gtm_white_box_test_case_number));
					repl_log(gtmrecv_log_fp, TRUE, TRUE, GTM_ZLIB_UNCMPLEN_ERROR_STR
						GTM_ZLIB_UNCMP_ERR_SEQNO_STR, destlen, gtmrecv_repl_uncmpmsglen,
						recv_jnl_seqno, recv_jnl_seqno);
					uncmpfail = TRUE;
				}
			}
		}
		if (uncmpfail)
		{	/* Since uncompression failed, default to NO compression. Send a REPL_CMP2UNCMP message accordingly */
			repl_log(gtmrecv_log_fp, TRUE, TRUE, GTM_ZLIB_UNCMPTRANSITION_STR);
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Waiting for update process to clear the backlog first\n");
			wait_for_updproc_to_clear_backlog();
			if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
				return;
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Update process has successfully cleared the backlog\n");
			gtmrecv_send_cmp2uncmp = TRUE;	/* trigger REPL_CMP2UNCMP message processing */
			gtmrecv_poll_actions(data_len, buff_unprocessed, buffp);
			assert(repl_connection_reset || (!gtmrecv_send_cmp2uncmp && gtmrecv_wait_for_jnl_seqno));
			return;
		}
		assert(0 == destlen % REPL_MSG_ALIGN);
		msgp = (repl_msg_ptr_t)gtmrecv_uncmpmsgp;
		msgp_top = (repl_msg_ptr_t)(gtmrecv_uncmpmsgp + destlen);
	} else
	{
		msgp = NULL;
		msgp_top = NULL;
	}
	do
	{
		assert(remote_side->endianness_known);	/* only then is remote_side->cross_endian reliable */
		if (is_repl_cmpc)
		{
			assert(msgp);
			assert(msgp_top);
			if (msgp >= msgp_top)
			{
				assert(msgp == msgp_top);
				break;
			}
			/* If primary is of different endianness, endian convert the UNCOMPRESSED message header
			 * before using the type and len fields (the compressed message header was already endian
			 * converted as part of receiving the message in do_main_loop())
			 */
			if (remote_side->cross_endian)
			{
				msgp->type = GTM_BYTESWAP_32(msgp->type);
				msgp->len = GTM_BYTESWAP_32(msgp->len);
			}
			assert(REPL_TR_JNL_RECS == msgp->type);
			cur_data_len = msgp->len - REPL_MSG_HDRLEN;
			assert(0 < cur_data_len);
			assert(0 == (cur_data_len % REPL_MSG_ALIGN));
			PREPARE_RECVPOOL_FOR_WRITE(cur_data_len, 0);	/* could update "recvpool_ctl->write" and "write_loc" */
			COPY_TO_RECVPOOL((uchar_ptr_t)msgp + REPL_MSG_HDRLEN, cur_data_len);/* uses and updates "write_loc" */
			msgp = (repl_msg_ptr_t)((uchar_ptr_t)msgp + cur_data_len + REPL_MSG_HDRLEN);
		}
		write_off = recvpool_ctl->write;
		write_len = (write_loc - write_off);
		assert((write_off != write_wrap) || (0 == write_off));
		assert(remote_side->jnl_ver);
		assert(!remote_side->cross_endian || (V18_JNL_VER <= remote_side->jnl_ver));
		if (ENDIAN_CONVERSION_NEEDED(is_new_histrec, this_side->jnl_ver, remote_side->jnl_ver, remote_side->cross_endian))
		{
			if (SS_NORMAL != (status = repl_tr_endian_convert(remote_side->jnl_ver,
							recvpool.recvdata_base + write_off, write_len)))
				RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(5) ERR_REPLXENDIANFAIL, 3, LEN_AND_LIT("Replicating"),
					&recvpool.upd_proc_local->read_jnl_seqno);
		}
		if (!is_new_histrec)
		{
			if (NO_FILTER != gtmrecv_filter)
			{	/* Need to pass through filter */
				pre_filter_write = write_off;
				pre_filter_write_len = write_len;
				/* If input transaction cannot fit in currently allocated buffer, expand buffer
				 * directly to that needed length hoping this should be enough (minimizes the # of
				 * 50% expansions we do). If this is still not enough (because the filter function
				 * results in bigger sized jnl records) we will then do geometric expansion of this
				 * buffer in the while loop below as much as needed.
				 */
				if (write_len > repl_filter_bufsiz)
				{
					gtmrecv_free_filter_buff();
					gtmrecv_alloc_filter_buff(write_len);
				}
				if (gtmrecv_filter & INTERNAL_FILTER)
				{
					in_buff = recvpool.recvdata_base + write_off;
					in_size = write_len;
					while (SS_NORMAL != (status =
						repl_filter_old2cur[remote_side->jnl_ver - JNL_VER_EARLIEST_REPL](
								in_buff, &in_size, repl_filter_buff, &out_size, repl_filter_bufsiz))
						&& (EREPL_INTLFILTER_NOSPC == repl_errno))
					{	/* We ran out of space in current buffer. We can try and use the transformed
						 * records as is and resume transformation from where the space-issue occurred
						 * but the transformation function might rely on seeing the entire transaction
						 * context in one shot so better not to take a risk. So free old buffer and
						 * allocate new buffer and redo transformation from scratch.
						 */
						gtmrecv_free_filter_buff();
						in_size = pre_filter_write_len;	/* just in case in_size was modified */
						write_len = write_len + (write_len >> 1); /* increase the buffer size by half */
						gtmrecv_alloc_filter_buff(write_len);
					}
					if (SS_NORMAL == status)
						write_len = out_size;
					else
					{
						if (EREPL_INTLFILTER_DATA2LONG == repl_errno)
							RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(4) ERR_JNLSETDATA2LONG, 2,
								jnl_source_datalen, jnl_dest_maxdatalen);
						else if (EREPL_INTLFILTER_NEWREC == repl_errno)
							RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(4) ERR_JNLNEWREC, 2,
								(unsigned int)jnl_source_rectype,
								(unsigned int)jnl_dest_maxrectype);
						else
						{
							INT_FILTER_RTS_ERROR(recvpool_ctl->jnl_seqno, repl_errno); /* no return */
						}
					}
				} else
					memcpy(repl_filter_buff, recvpool.recvdata_base + write_off, write_len);
				assert(write_len <= repl_filter_bufsiz);
				GTMTRIG_ONLY(
					if ((unsigned char)V19_JNL_VER <= remote_side->jnl_ver)
					{
						repl_sort_tr_buff(repl_filter_buff, write_len);
						DBG_VERIFY_TR_BUFF_SORTED(repl_filter_buff, write_len);
					}
				)
				if ((gtmrecv_filter & EXTERNAL_FILTER)
					&& (SS_NORMAL !=
						(status = repl_filter(recvpool_ctl->jnl_seqno, &repl_filter_buff,
											(int*)&write_len, &repl_filter_bufsiz))))
					repl_filter_error(recvpool_ctl->jnl_seqno, status);
				GTMTRIG_ONLY(
					/* Ensure that the external filter has not disturbed the sorted sequence of the
					 * update_num
					 */
					DEBUG_ONLY(
						if ((unsigned char)V19_JNL_VER <= remote_side->jnl_ver)
							DBG_VERIFY_TR_BUFF_SORTED(repl_filter_buff, write_len);
					)
				)
				assert(write_len <= repl_filter_bufsiz);
				write_loc = write_off;		/* reset "write_loc" */
				PREPARE_RECVPOOL_FOR_WRITE(write_len, pre_filter_write_len);	/* could update "->write"
												 * and "write_loc" */
				COPY_TO_RECVPOOL((uchar_ptr_t)repl_filter_buff, write_len);/* uses and updates "write_loc" */
				write_off = recvpool_ctl->write;
				repl_recv_postfltr_data_procd += (qw_num)write_len;
				filter_pass = TRUE;
			} else
			{
				GTMTRIG_ONLY(
					if ((unsigned char)V19_JNL_VER <= remote_side->jnl_ver)
					{
						repl_sort_tr_buff((uchar_ptr_t)(recvpool.recvdata_base + write_off), write_len);
						DBG_VERIFY_TR_BUFF_SORTED((recvpool.recvdata_base + write_off), write_len);
					}
				)
			}
		}
		if (recvpool_ctl->jnl_seqno - lastlog_seqno >= log_interval)
		{
			log_seqno = recvpool_ctl->jnl_seqno;
			upd_seqno = recvpool.upd_proc_local->read_jnl_seqno;
			assert(log_seqno >= upd_seqno);
			diff_seqno = (log_seqno - upd_seqno);
			trans_recvd_cnt += (log_seqno - lastlog_seqno);
			msg_total = repl_recv_data_recvd - buff_unprocessed;
				/* Don't include data not yet processed, we'll include that count in a later log */
			if (NO_FILTER == gtmrecv_filter)
			{
				repl_log(gtmrecv_log_fp, TRUE, TRUE, "REPL INFO - Seqno : "INT8_FMT" "INT8_FMTX
					"  Jnl Total : "INT8_FMT" "INT8_FMTX"  Msg Total : "INT8_FMT" "INT8_FMTX
					"  Current backlog : "INT8_FMT" "INT8_FMTX"\n",
					log_seqno, log_seqno, repl_recv_data_processed, repl_recv_data_processed,
					msg_total, msg_total, diff_seqno, diff_seqno);
			} else
			{
				repl_log(gtmrecv_log_fp, TRUE, TRUE, "REPL INFO - Seqno : "INT8_FMT" "INT8_FMTX
					"  Pre filter total : "INT8_FMT" "INT8_FMTX"  Post filter total : "
					INT8_FMT" "INT8_FMTX"  Msg Total : "INT8_FMT" "INT8_FMTX
					"  Current backlog : "INT8_FMT" "INT8_FMTX"\n",
					log_seqno, log_seqno,
					repl_recv_data_processed, repl_recv_data_processed,
					repl_recv_postfltr_data_procd, repl_recv_postfltr_data_procd,
					msg_total, msg_total, diff_seqno, diff_seqno);
			}
			/* Approximate time with an error not more than GTMRECV_HEARTBEAT_PERIOD. We use this
			 * instead of calling time(), and expensive system call, especially on VMS. The
			 * consequence of this choice is that we may defer logging when we may have logged. We
			 * can live with that. Currently, the logging interval is not changeable by users.
			 * When/if we provide means of choosing log interval, this code may have to be re-examined.
			 * 	- Vinaya 2003/09/08.
			 */
			assert(0 != gtmrecv_now);
			repl_recv_this_log_time = gtmrecv_now;
			assert(repl_recv_this_log_time >= repl_recv_prev_log_time);
			time_elapsed = difftime(repl_recv_this_log_time, repl_recv_prev_log_time);
			if ((double)GTMRECV_LOGSTATS_INTERVAL <= time_elapsed)
			{
				repl_log(gtmrecv_log_fp, TRUE, FALSE, "REPL INFO since last log : Time elapsed : "
					"%00.f  Tr recvd : "INT8_FMT"  Tr bytes : "INT8_FMT"  Msg bytes : "INT8_FMT"\n",
					time_elapsed, trans_recvd_cnt - last_log_tr_recvd_cnt,
					repl_recv_data_processed - repl_recv_lastlog_data_procd,
					msg_total - repl_recv_lastlog_data_recvd);
				repl_log(gtmrecv_log_fp, TRUE, TRUE, "REPL INFO since last log : Time elapsed : "
					"%00.f  Tr recvd/s : %f  Tr bytes/s : %f  Msg bytes/s : %f\n", time_elapsed,
					(float)(trans_recvd_cnt - last_log_tr_recvd_cnt)/time_elapsed,
					(float)(repl_recv_data_processed - repl_recv_lastlog_data_procd)/time_elapsed,
					(float)(msg_total - repl_recv_lastlog_data_recvd)/time_elapsed);
				repl_recv_lastlog_data_procd = repl_recv_data_processed;
				repl_recv_lastlog_data_recvd = msg_total;
				last_log_tr_recvd_cnt = trans_recvd_cnt;
				repl_recv_prev_log_time = repl_recv_this_log_time;
			}
			lastlog_seqno = log_seqno;
		}
		if (gtmrecv_logstats)
		{
			if (!filter_pass)
			{
				repl_log(gtmrecv_log_fp, FALSE, FALSE, "Tr : "INT8_FMT"  Size : %d  Write : %ld  "
					 "Total : "INT8_FMT"\n", recvpool_ctl->jnl_seqno, write_len,
					 write_off, repl_recv_data_processed);
			} else
			{
				assert(!is_new_histrec);
				repl_log(gtmrecv_log_fp, FALSE, FALSE, "Tr : "INT8_FMT"  Pre filter Size : %d  "
					"Post filter Size  : %d  Pre filter Write : %d  Post filter Write : %ld  "
					"Pre filter Total : "INT8_FMT"  Post filter Total : "INT8_FMT"\n",
					recvpool_ctl->jnl_seqno, pre_filter_write_len, write_len,
					pre_filter_write, write_off, repl_recv_data_processed,
					repl_recv_postfltr_data_procd);
			}
		}
		recvpool_ctl->write_wrap = write_wrap;
		if (recvpool.gtmrecv_local->noresync)
		{	/* With -NORESYNC, we could take the strm_seqno further back with multiple connects of the same receiver
			 * server with different source servers at different seqnos. So avoid any confusion with asserts
			 * that check for increasing seqnos using last_valid_histinfo and last_rcvd_histinfo.
			 */
			memset(&recvpool.recvpool_ctl->last_valid_histinfo, 0, SIZEOF(recvpool.recvpool_ctl->last_valid_histinfo));
			memset(&recvpool.recvpool_ctl->last_rcvd_histinfo, 0, SIZEOF(recvpool.recvpool_ctl->last_rcvd_histinfo));
			/* In the case of -NORESYNC, we are a root primary supplementary instance (not a propagating primary)
			 * and therefore recvpool.recvpool_ctl->last_valid_strm_histinfo[] and
			 * recvpool.recvpool_ctl->last_rcvd_strm_histinfo[] are unused and hence no need to reset them.
			 */
			assert(!remote_side->is_supplementary);	/* assert that we are not a propagating primary supplementary */
			assert(jnlpool->repl_inst_filehdr->is_supplementary && !jnlpool->jnlpool_ctl->upd_disabled);
		}
		if (is_new_histrec)
		{
			/* Note: The REPL_OLD_TRIPLE or REPL_HISTREC messages are endian converted by the source server
			 * in case the receiver is running with a this_side->jnl_ver higher than the source. If not, the call to
			 * function "repl_tr_endian_convert" (done already in the current function) will take care of
			 * the endian conversion. So no more endian conversion needed at this point.
			 */
			if (REPL_OLD_TRIPLE == msg_type)
			{
				assert(REPL_PROTO_VER_MULTISITE <= remote_side->proto_ver);
				assert(REPL_PROTO_VER_SUPPLEMENTARY > remote_side->proto_ver);
				assert(SIZEOF(old_triple_content) == write_len);
				memcpy((sm_uc_ptr_t)&old_triple_content, (recvpool.recvdata_base + write_off),
					SIZEOF(old_triple_content));
				assert(JRT_TRIPLE == old_triple_content.jrec_type);
				assert(old_triple_content.forwptr == SIZEOF(old_triple_content));
				assert(old_triple_content.start_seqno == recvpool_ctl->jnl_seqno);
				assert(old_triple_content.start_seqno >= recvpool.upd_proc_local->read_jnl_seqno);
				assert((old_triple_content.start_seqno > recvpool_ctl->last_valid_histinfo.start_seqno)
					 || ((old_triple_content.start_seqno == recvpool_ctl->last_valid_histinfo.start_seqno)
						&& gtm_white_box_test_case_enabled
						&& (WBTEST_UPD_PROCESS_ERROR == gtm_white_box_test_case_number)));
				cur_histinfo = &tmp_histjrec.histcontent;
				memcpy(cur_histinfo->root_primary_instname, old_triple_content.instname, MAX_INSTNAME_LEN - 1);
				cur_histinfo->root_primary_instname[MAX_INSTNAME_LEN - 1] = '\0';
				cur_histinfo->start_seqno = old_triple_content.start_seqno;
				cur_histinfo->strm_seqno = 0;
				cur_histinfo->root_primary_cycle = old_triple_content.cycle;
				cur_histinfo->creator_pid = 0;
				cur_histinfo->created_time = 0;
				/* No need to initialize the following fields as they are reinitialized later
				 * when the history record gets added to the instance file (in "repl_inst_histinfo_add").
				 *	cur_histinfo->histinfo_num = INVALID_HISTINFO_NUM;
				 *	cur_histinfo->prev_histinfo_num = INVALID_HISTINFO_NUM;
				 *	cur_histinfo->last_histinfo_num[] = INVALID_HISTINFO_NUM;
				 */
				cur_histinfo->strm_index = 0;
				cur_histinfo->history_type = HISTINFO_TYPE_NORMAL;
				NULL_INITIALIZE_REPL_INST_UUID(cur_histinfo->lms_group);
				tmp_histjrec.jrec_type = JRT_HISTREC;
				tmp_histjrec.forwptr = SIZEOF(repl_histrec_jnl_t);
				/* We now have to upgrade a REPL_OLD_TRIPLE format history record to a REPL_HISTREC record.
				 * The latter is a much bigger record so make room for this in the receive pool.
				 */
				write_loc = write_off;		/* reset "write_loc" */
				write_len = tmp_histjrec.forwptr;
				PREPARE_RECVPOOL_FOR_WRITE(write_len, SIZEOF(old_triple_content));
					/* above macro could update "recvpool_ctl->write" and "write_loc" */
				COPY_TO_RECVPOOL((uchar_ptr_t)&tmp_histjrec, write_len);/* uses and updates "write_loc" */
				write_off = recvpool_ctl->write;
				/* Copy relevant fields from received histinfo message to "last_rcvd_histinfo" if applicable */
				cur_histinfo = &recvpool_ctl->last_rcvd_histinfo;
				assert(old_triple_content.start_seqno >= cur_histinfo->start_seqno);
				*cur_histinfo = tmp_histjrec.histcontent;
				assert(!remote_side->is_supplementary); /* so no need to maintain last_rcvd_strm_histinfo[] */
			} else
			{
				assert(REPL_HISTREC == msg_type);
				assert(REPL_PROTO_VER_SUPPLEMENTARY <= remote_side->proto_ver);
				assert(SIZEOF(repl_histrec_jnl_t) == write_len);
				pool_histrec = (repl_histrec_jnl_t *)((sm_uc_ptr_t)recvpool.recvdata_base + write_off);
				assert(JRT_HISTREC == pool_histrec->jrec_type);
				assert(pool_histrec->forwptr == SIZEOF(repl_histrec_jnl_t));
				pool_histinfo = &pool_histrec->histcontent;
				assert(pool_histinfo->start_seqno == recvpool_ctl->jnl_seqno);
				assert(pool_histinfo->start_seqno >= recvpool.upd_proc_local->read_jnl_seqno);
				if (jnlpool->repl_inst_filehdr->is_supplementary && !jnlpool->jnlpool_ctl->upd_disabled)
				{	/* Modify the history record to reflect the stream # */
					assert((0 <= recvpool.gtmrecv_local->strm_index)
						&& (MAX_SUPPL_STRMS > recvpool.gtmrecv_local->strm_index));
					assert(strm_index == recvpool.gtmrecv_local->strm_index);
					pool_histinfo->strm_index = recvpool.gtmrecv_local->strm_index;
					assert(0 == pool_histinfo->strm_seqno);
					assert(IS_REPL_INST_UUID_NON_NULL(recvpool.gtmrecv_local->remote_lms_group));
					pool_histinfo->lms_group = recvpool.gtmrecv_local->remote_lms_group;
					if (recvpool.gtmrecv_local->updateresync)
					{
						assert(FD_INVALID != recvpool.gtmrecv_local->updresync_instfile_fd);
						/* Make it known that this is an updateresync type history record */
						pool_histinfo->history_type = HISTINFO_TYPE_UPDRESYNC;
					}
					/* We want to do a similar check for -noresync but we cannot use gtmrecv_local->noresync
					 * since that is reset the moment a REPL_WILL_RESTART_WITH_INFO message is seen (much
					 * before coming here). So use a static variable to determine if this is the first
					 * history record that is being received by this receiver server and if -noresync was
					 * specified in the command line (indicated by gtmrecv_options.noresync), set the
					 * history_type of the incoming history record to reflect the noresync type.
					 */
					assert(!recvpool.gtmrecv_local->noresync);
					if (first_histrec)
					{
						if (gtmrecv_options.noresync)
							pool_histinfo->history_type = HISTINFO_TYPE_NORESYNC;
						first_histrec = FALSE;
					}
				}
				assert((INVALID_SUPPL_STRM != strm_index) || (0 == pool_histinfo->strm_index));
				cur_histinfo = &recvpool_ctl->last_rcvd_histinfo;
				*cur_histinfo = *pool_histinfo;
				if ((INVALID_SUPPL_STRM == strm_index) || (strm_index == pool_histinfo->strm_index))
				{
					assert((pool_histinfo->start_seqno > recvpool_ctl->last_valid_histinfo.start_seqno)
						 || ((pool_histinfo->start_seqno == recvpool_ctl->last_valid_histinfo.start_seqno)
							&& gtm_white_box_test_case_enabled
							&& (WBTEST_UPD_PROCESS_ERROR == gtm_white_box_test_case_number)));
					assert(pool_histinfo->start_seqno >= cur_histinfo->start_seqno);
				} else
				{
					assert(!recvpool_ctl->insert_strm_histinfo);
					cur_histinfo = pool_histinfo;
				}
			}
			/* If supplementary instance with updates disabled, initialize last_rcvd_strm_histinfo as well */
			if (remote_side->is_supplementary)
			{	/* Check if "insert_strm_histinfo" is set. If so and if REPL_STRMINFO messages were
				 * exchanged between the source and receiver, we need to potentially insert history
				 * records for valid stream #s > 0. This is to ensure that even in case of a
				 * -updateresync startup, we record knowledge of all streams that exist on the source side
				 * (which we previously received as part of the same connection) on the receiver side as well.
				 * Examples of when REPL_STRMINFO messages are not exchanged are if receiver side is at
				 * jnl_seqno = 1, or if the receiver side is at the exact same jnl_seqno as the source side
				 * and no prior history is available.
				 */
				if (recvpool_ctl->insert_strm_histinfo
					&& (max_strm_histinfo = recvpool_ctl->max_strm_histinfo)) /* caution: assignment */
				{	/* Make room for the new records in the receive pool first.
					 * At this point "cur_histinfo" holds the first received history record.
					 * Add the other stream history records onto it.
					 */
					assert(0 == cur_histinfo->strm_index);
					/* Determine how many history records need to be added */
					num_strm_histinfo = 0;
					/* Copy 0th stream history record into receive pool first */
					rcvd_strm_histjrec[num_strm_histinfo].jrec_type = JRT_HISTREC;
					rcvd_strm_histjrec[num_strm_histinfo].forwptr = SIZEOF(repl_histrec_jnl_t);
					cur_histinfo->strm_seqno = jnlpool->jnlpool_ctl->strm_seqno[0];
					rcvd_strm_histjrec[num_strm_histinfo++].histcontent = *cur_histinfo;
					/* Copy non-zero stream history records if they exist */
					for (idx = 1; idx < max_strm_histinfo; idx++)
					{
						if (recvpool_ctl->is_valid_strm_histinfo[idx])
						{
							DEBUG_ONLY(tmp_histinfo = recvpool_ctl->last_rcvd_strm_histinfo[idx]);
							assert(IS_REPL_INST_UUID_NON_NULL(tmp_histinfo.lms_group));
							rcvd_strm_histjrec[num_strm_histinfo].jrec_type = JRT_HISTREC;
							rcvd_strm_histjrec[num_strm_histinfo].forwptr = SIZEOF(repl_histrec_jnl_t);
							rcvd_strm_histjrec[num_strm_histinfo].histcontent
								= recvpool_ctl->last_rcvd_strm_histinfo[idx];
							/* Fix the "start_seqno" & "strm_seqno" fields of the history record */
							rcvd_strm_histjrec[num_strm_histinfo].histcontent.start_seqno
								= cur_histinfo->start_seqno;
							/* Note that jnlpool_ctl->strm_seqno[idx] could be 0 if we have not yet
							 * seen any updates in this stream. But since we do have a history record
							 * for this stream and its strm_seqno cannot be 0 (has to be non-zero),
							 * set it to 1 in that case.
							 */
							rcvd_strm_histjrec[num_strm_histinfo].histcontent.strm_seqno
								= jnlpool->jnlpool_ctl->strm_seqno[idx]
									? jnlpool->jnlpool_ctl->strm_seqno[idx] : 1;
							num_strm_histinfo++;
							/* Do not reset recvpool_ctl->is_valid_strm_histinfo[idx] to FALSE
							 * as this reflects reality and staying this way helps later
							 * communication with the source in cases of reconnects/pipe-drains.
							 */
						}
					}
					write_loc = write_off;		/* reset "write_loc" */
					write_len = num_strm_histinfo * SIZEOF(repl_histrec_jnl_t);
					PREPARE_RECVPOOL_FOR_WRITE(write_len, 0);
						/* could update "recvpool_ctl->write" * and "write_loc" */
					COPY_TO_RECVPOOL((uchar_ptr_t)rcvd_strm_histjrec, write_len);
						/* uses and updates "write_loc" */
					write_off = recvpool_ctl->write;
					/* Note: "last_rcvd_strm_histinfo" is automatically maintained in this case */
					/* Now that we have inserted the stream specific history records, reset flag */
					recvpool_ctl->insert_strm_histinfo = FALSE;
				} else
				{	/* Maintain last_rcvd_strm_histinfo[] as well.
					 * In addition, if insert_strm_histinfo is TRUE, it means no REPL_STRMINFO messages
					 * were exchanged at startup. In that case, no need of inserting stream specific
					 * history information in the receive pool so reset variable to FALSE.
					 */
					if (recvpool_ctl->insert_strm_histinfo)
						recvpool_ctl->insert_strm_histinfo = FALSE;
					histinfo_strm_num = cur_histinfo->strm_index;
					assert((0 <= histinfo_strm_num) && (MAX_SUPPL_STRMS > histinfo_strm_num));
					memcpy(&recvpool_ctl->last_rcvd_strm_histinfo[histinfo_strm_num],
							cur_histinfo, SIZEOF(repl_histinfo));
					recvpool_ctl->is_valid_strm_histinfo[histinfo_strm_num] = TRUE;
					if (recvpool_ctl->max_strm_histinfo <= histinfo_strm_num)
						recvpool_ctl->max_strm_histinfo = histinfo_strm_num + 1;
				}
			}
			repl_dump_histinfo(gtmrecv_log_fp, TRUE, TRUE, "New History Content", cur_histinfo);
		} else
		{	/* Note: In case of a propagating primary supplementary instance, the below if implies that we will
			 * change last_rcvd_histinfo to last_valid_histinfo even if the logical update corresponded to a
			 * different stream than 0 (the stream corresponding to the history of interest on this receiver).
			 * But there should not be any issues due to this as it is an update nevertheless and is going to
			 * bump up the recvpool_ctl->jnl_seqno to one more than last_rcvd_histinfo.start_seqno.
			 */
			if (recvpool_ctl->jnl_seqno == recvpool_ctl->last_rcvd_histinfo.start_seqno)
			{	/* Move over stuff from "last_rcvd_histinfo" to "last_valid_histinfo" */
				memcpy(&recvpool_ctl->last_valid_histinfo,
					&recvpool_ctl->last_rcvd_histinfo, SIZEOF(repl_histinfo));
				if (remote_side->is_supplementary)
				{	/* Propagating primary supplementary instance. Maintain last_valid_strm_histinfo too. */
					max_strm_histinfo = recvpool_ctl->max_strm_histinfo;
					assert(max_strm_histinfo);
					for (idx = 0; idx < max_strm_histinfo; idx++)
					{
						if (recvpool_ctl->is_valid_strm_histinfo[idx])
						{
							DEBUG_ONLY(tmp_histinfo = recvpool_ctl->last_rcvd_strm_histinfo[idx]);
							assert((0 == idx) || IS_REPL_INST_UUID_NON_NULL(tmp_histinfo.lms_group));
							assert((0 != idx) || IS_REPL_INST_UUID_NULL(tmp_histinfo.lms_group));
							memcpy(&recvpool_ctl->last_valid_strm_histinfo[idx],
								&recvpool_ctl->last_rcvd_strm_histinfo[idx], SIZEOF(repl_histinfo));
							recvpool_ctl->is_valid_strm_histinfo[idx] = FALSE;
						}
					}
					recvpool_ctl->max_strm_histinfo = 0;
					/* Now that last_valid_strm_histinfo[0] is initialized, it is safe to reset the below */
					if (recvpool_ctl->insert_strm_histinfo)
						recvpool_ctl->insert_strm_histinfo = FALSE;
				}
				/* Now that at least one history record has been written into the receive pool and is guaranteed
				 * to be written to the instance file (when this gets processed by the update process), don't use
				 * -updateresync or -noresync for future handshakes in case the current connection gets reset.
				 */
				if (recvpool.gtmrecv_local->updateresync)
				{
					recvpool.gtmrecv_local->updateresync = FALSE;
					/* Close fd of the input instance file name used for the -updateresync */
					if (FD_INVALID != recvpool.gtmrecv_local->updresync_instfile_fd)
						CLOSEFILE_RESET(recvpool.gtmrecv_local->updresync_instfile_fd, rc);
				}
				assert(!recvpool.gtmrecv_local->noresync);	/* should have been reset already */
			}
			QWINCRBYDW(recvpool_ctl->jnl_seqno, 1);
			assert(recvpool_ctl->last_valid_histinfo.start_seqno < recvpool_ctl->jnl_seqno);
		}
		/* The update process looks at "recvpool_ctl->write" first and then reads (a) "recvpool_ctl->write_wrap"
		 * AND (b) all journal data in the receive pool upto this offset. It assumes that (a) and (b) will never
		 * hold stale values corresponding to a previous state of "recvpool_ctl->write". In order for this
		 * assumption to hold good, the receiver server needs to do a write memory barrier after updating the
		 * receive pool data and "write_wrap" but before updating "write". The update process will do a read
		 * memory barrier after reading "write" but before reading "write_wrap" or the receive pool data. Not
		 * enforcing the read order will result in the update process attempting to read/process invalid data
		 * from the receive pool (which could end up in db out of sync situation between primary and secondary).
		 */
		SHM_WRITE_MEMORY_BARRIER;
		recvpool_ctl->write = write_loc;
		/* Signal the update process to check for the update. */
		PTHREAD_COND_SIGNAL(&recvpool_ctl->write_updated, status);
		if (0 != status)
			RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(8) ERR_SYSCALL, 5,
				LEN_AND_LIT("pthread_cond_signal"), CALLFROM, status, 0);
	} while (is_repl_cmpc);
	return;
}

/* Retrieve the history record corresponding to "input_seqno" in the instance file used with the -UPDATERESYNC qualifier.
 * Do special processing in case of P->Q type of connection (where primary and secondary are both supplementary instances)
 * which is to search only in the 0th stream.
 *
 * Note: This function is similar to "repl_inst_histinfo_find_seqno" except that this operates on the input instance file
 * provided with the -updateresync qualifier. Two reasons why we need this code duplication is
 *	a) The "repl_inst_histinfo_find_seqno" function currently is coded to use only the replication instance file.
 *	b) The input instance file for -updateresync could be cross endian.
 * If "repl_inst_histinfo_find_seqno" is enhanced to fix these limitations, then we can avoid this code duplication.
 */
STATICFNDEF void	gtmrecv_updresync_histinfo_find_seqno(seq_num input_seqno, int4 strm_num, repl_histinfo *histinfo)
{
	char		print_msg[1024];
	int		fd, status;
	int4		histinfo_num;
	off_t		offset;

	fd = recvpool.gtmrecv_local->updresync_instfile_fd;
	assert(FD_INVALID != fd);
	/* If remote side is non-supplementary then its instance file (which is given as input to the -updateresync
	 * command) knows only strem 0, so reset strm_num to 0 unconditionally.
	 */
	if (!remote_side->is_supplementary)
		strm_num = 0;
	if (INVALID_SUPPL_STRM == strm_num)
		histinfo_num = recvpool.gtmrecv_local->updresync_num_histinfo;
	else
		histinfo_num = recvpool.gtmrecv_local->updresync_num_histinfo_strm[strm_num];
	if (INVALID_HISTINFO_NUM == histinfo_num)
	{	/* The instance file cannot be used for updateresync if it has NO history records. */
		RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(9) ERR_UPDSYNCINSTFILE, 0, ERR_STRMNUMIS, 1, strm_num, ERR_TEXT, 2,
			LEN_AND_LIT("Input instance file has NO history records"));
	}
	assert(0 <= histinfo_num);
	if (!recvpool.gtmrecv_local->updresync_jnl_seqno)
	{	/* The instance file cannot be used for updateresync if it has a ZERO seqno */
		RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(9) ERR_UPDSYNCINSTFILE, 0, ERR_STRMNUMIS, 1, strm_num,
			ERR_TEXT, 2, LEN_AND_LIT("Input instance file has jnl_seqno of 0"));
	}
	if (input_seqno > recvpool.gtmrecv_local->updresync_jnl_seqno)
	{	/* Input seqno is greater than the max seqno in the updateresync input instance file. So can never be found. */
		SNPRINTF(print_msg, SIZEOF(print_msg), "Seqno "INT8_FMT" "INT8_FMTX" cannot be found in input instance file "
			" which has a max seqno of "INT8_FMT" "INT8_FMTX"\n", input_seqno, input_seqno,
			recvpool.gtmrecv_local->updresync_jnl_seqno, recvpool.gtmrecv_local->updresync_jnl_seqno);
		RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(9) ERR_UPDSYNCINSTFILE, 0,
			ERR_STRMNUMIS, 1, strm_num, ERR_TEXT, 2, LEN_AND_STR(print_msg));
	}
	histinfo->start_seqno = 0;
	do
	{
		offset = REPL_INST_HISTINFO_START + ((histinfo_num) * SIZEOF(repl_histinfo));
		LSEEKREAD(fd, offset, histinfo, SIZEOF(repl_histinfo), status);
		if (0 != status)
		{	/* At this point, we don't have the name of the input instance file used in the -updateresync qualifier.
			 * So we use a value of "" instead. The fact that the REPLINSTREAD message is preceded by a UPDSYNCINSTFILE
			 * error indicates to the user it is the -updateresync qualifier where the issue is so it is not a big loss.
			 */
			if (-1 == status)
				RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(15) ERR_UPDSYNCINSTFILE, 0,
					ERR_STRMNUMIS, 1, strm_num,
					ERR_TEXT, 2, LEN_AND_LIT("Error reading history record"),
					ERR_REPLINSTREAD, 4, SIZEOF(repl_histinfo), (qw_off_t *)&offset, LEN_AND_LIT(""));
			else
				RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(16) ERR_UPDSYNCINSTFILE, 0,
					ERR_STRMNUMIS, 1, strm_num,
					ERR_TEXT, 2, LEN_AND_LIT("Error reading history record"),
					ERR_REPLINSTREAD, 4, SIZEOF(repl_histinfo), (qw_off_t *)&offset, LEN_AND_LIT(""), status);
		}
		if (recvpool.gtmrecv_local->updresync_cross_endian)
			ENDIAN_CONVERT_REPL_HISTINFO(histinfo);
		if (input_seqno > histinfo->start_seqno)
			return;		/* found history record corresponding to input_seqno. return right away */
		histinfo_num = (INVALID_SUPPL_STRM == strm_num) ? (histinfo_num - 1) : histinfo->prev_histinfo_num;
	} while (INVALID_HISTINFO_NUM != histinfo_num);
	/* Could not find history record in -updateresync= input instance file */
	SNPRINTF(print_msg, SIZEOF(print_msg),
			"Receiver side instance seqno "INT8_FMT" "INT8_FMTX" is less than"
			" any history record found in instance file", input_seqno, input_seqno);
	RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(9) ERR_UPDSYNCINSTFILE, 0,
		ERR_STRMNUMIS, 1, strm_num, ERR_TEXT, 2, LEN_AND_STR(print_msg));
}

/* Retrieve the "index"'th history record in the instance file specified using the -UPDATERESYNC qualifier.
 *
 * This function is similar to "repl_inst_histinfo_get" except that this operates on the input instance file provided
 * with the -updateresync qualifier. Two reasons why we need this code duplication is
 *	a) The "repl_inst_histinfo_get" function currently is coded to use only the replication instance file.
 *	b) The input instance file for -updateresync could be cross endian.
 * If "repl_inst_histinfo_get" is enhanced to fix these limitations, then we can avoid this code duplication.
 */
STATICFNDEF void	gtmrecv_updresync_histinfo_get(int4 index, repl_histinfo *histinfo)
{
	int		fd, status;
	off_t		offset;

	fd = recvpool.gtmrecv_local->updresync_instfile_fd;
	assert(FD_INVALID != fd);
	assert(0 <= index);
	offset = REPL_INST_HISTINFO_START + ((index) * SIZEOF(repl_histinfo));
	LSEEKREAD(fd, offset, histinfo, SIZEOF(repl_histinfo), status);
	if (0 != status)
	{	/* At this point, we don't have the name of the input instance file used in the -updateresync qualifier.
		 * So we use a value of "" instead. The fact that the REPLINSTREAD message is preceded by a UPDSYNCINSTFILE
		 * error indicates to the user it is the -updateresync qualifier where the issue is so it is not a big loss.
		 */
		if (-1 == status)
			RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(12) ERR_UPDSYNCINSTFILE, 0,
				ERR_TEXT, 2, LEN_AND_LIT("Error reading history record"),
				ERR_REPLINSTREAD, 4, SIZEOF(repl_histinfo), (qw_off_t *)&offset, LEN_AND_LIT(""));
		else
			RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(13) ERR_UPDSYNCINSTFILE, 0,
				ERR_TEXT, 2, LEN_AND_LIT("Error reading history record"),
				ERR_REPLINSTREAD, 4, SIZEOF(repl_histinfo), (qw_off_t *)&offset, LEN_AND_LIT(""), status);
	}
	if (recvpool.gtmrecv_local->updresync_cross_endian)
		ENDIAN_CONVERT_REPL_HISTINFO(histinfo);
}

/* This function is invoked on receipt of a REPL_NEED_STRMINFO message. In case of an error, it returns if either
 * repl_connection_reset OR gtmrecv_wait_for_jnl_seqno are set so the caller should check for this and return as well.
 */
STATICFNDEF void	gtmrecv_process_need_strminfo_msg(repl_needstrminfo_msg_ptr_t need_strminfo_msg)
{
	int4			status;
	int			idx;
	repl_histinfo		histinfo, *previously_rcvd_histinfo;
	repl_strminfo_msg_t	strminfo_msg;
	seq_num			need_strminfo_seqno, last_valid_histinfo_seqno;

	assert(remote_side->is_supplementary);	/* STRMINFO messages are sent only in case source and receiver are supplementary */
	assert(0 == strm_index);
	assert(remote_side->endianness_known); /* ensure remote_side->cross_endian is reliable */
	if (!remote_side->cross_endian)
		need_strminfo_seqno = need_strminfo_msg->seqno;
	else
		need_strminfo_seqno = GTM_BYTESWAP_64(need_strminfo_msg->seqno);
	repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received REPL_NEED_STRMINFO message for seqno "INT8_FMT" "INT8_FMTX"\n",
			need_strminfo_seqno, need_strminfo_seqno);
	if (recvpool.gtmrecv_local->updateresync && (FD_INVALID != recvpool.gtmrecv_local->updresync_instfile_fd))
	{	/* The stream information that is being requested needs to be found in the -updateresync input instance file
		 * (not the receiver side instance file).
		 */
		repl_log(gtmrecv_log_fp, TRUE, TRUE, "Searching for desired stream info in -updateresync input instance file\n");
		gtmrecv_updresync_histinfo_find_seqno(need_strminfo_seqno, INVALID_SUPPL_STRM, &histinfo);
		for (idx = 0; idx < MAX_SUPPL_STRMS; idx++)
			strminfo_msg.last_histinfo_num[idx] = histinfo.last_histinfo_num[idx];
		assert(recvpool.recvpool_ctl->insert_strm_histinfo);
	} else
	{	/* The history record needs to be found in the receiver side instance file or in the receive pool.
		 * If last_valid_strm_histinfo[0] has non-default content, then because this is a supplementary instance
		 * and a propagating primary, we can rest assured that last_valid_strm_histinfo[1] thru [15] reflect
		 * the latest history records for each stream if the stream exists on this instance. And so no need
		 * to go to the instance file at all. If [0] does not have any content, then it means the cached history
		 * is empty for not just the 0th stream but for every other stream as well i.e. the receive pool is empty
		 * and the receiver is connecting with a source for the first time. So go to the instance file in that case.
		 */
		last_valid_histinfo_seqno = recvpool.recvpool_ctl->last_valid_strm_histinfo[0].start_seqno;
		if (last_valid_histinfo_seqno)
		{
			assert(need_strminfo_seqno > last_valid_histinfo_seqno);
			assert(!recvpool.gtmrecv_local->updateresync);
			assert(!recvpool.gtmrecv_local->noresync);
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Searching for the desired history in the receive pool\n");
			for (idx = 0; idx < MAX_SUPPL_STRMS; idx++)
			{
				previously_rcvd_histinfo = &recvpool.recvpool_ctl->last_valid_strm_histinfo[idx];
				assert((0 != idx) || IS_REPL_INST_UUID_NULL(previously_rcvd_histinfo->lms_group));
				if ((0 == idx) || IS_REPL_INST_UUID_NON_NULL(previously_rcvd_histinfo->lms_group))
					strminfo_msg.last_histinfo_num[idx] = UNKNOWN_HISTINFO_NUM;
				else
				{
					assert(0 != idx);
					strminfo_msg.last_histinfo_num[idx] = INVALID_HISTINFO_NUM;
				}
			}
			assert(!recvpool.recvpool_ctl->insert_strm_histinfo);
			histinfo.strm_index = 0;	/* so "0 < histinfo.strm_index" if block is skipped below for this case */
		} else
		{
			repl_log(gtmrecv_log_fp, TRUE, TRUE,
				"Searching for the desired history in the replication instance file\n");
			grab_lock(jnlpool->jnlpool_dummy_reg, TRUE, GRAB_LOCK_ONLY);
			GTMRECV_ONLN_RLBK_CLNUP_IF_NEEDED;
				/* above macro will "return" if repl_connection_reset OR gtmrecv_wait_for_jnl_seqno is set */
			status = repl_inst_wrapper_histinfo_find_seqno(need_strminfo_seqno, INVALID_SUPPL_STRM, &histinfo);
			rel_lock(jnlpool->jnlpool_dummy_reg);
			if (0 != status)
			{	/* Close the connection */
				assert(ERR_REPLINSTNOHIST == status);
				assert(FALSE);
				gtmrecv_autoshutdown();	/* should not return */
			}
			for (idx = 0; idx < MAX_SUPPL_STRMS; idx++)
				strminfo_msg.last_histinfo_num[idx] = histinfo.last_histinfo_num[idx];
		}
	}
	if (0 < histinfo.strm_index)
	{
		assert(histinfo.last_histinfo_num[histinfo.strm_index] < histinfo.histinfo_num);
		strminfo_msg.last_histinfo_num[histinfo.strm_index] = histinfo.histinfo_num;
	}
	if (remote_side->cross_endian)
	{
		for (idx = 0; idx < MAX_SUPPL_STRMS; idx++)
		{
			assert(4 == SIZEOF(strminfo_msg.last_histinfo_num[idx]));	/* so GTM_BYTESWAP_32 can be used below */
			strminfo_msg.last_histinfo_num[idx] = GTM_BYTESWAP_32(strminfo_msg.last_histinfo_num[idx]);
		}
	}
	gtmrecv_repl_send((repl_msg_ptr_t)&strminfo_msg, REPL_STRMINFO, SIZEOF(repl_strminfo_msg_t), "REPL_STRMINFO", MAX_SEQNO);
	return;
}

/* This function is invoked on receipt of a REPL_NEED_HISTINFO message. In case of an error, it returns if either
 * repl_connection_reset OR gtmrecv_wait_for_jnl_seqno are set so the caller should check for this and return as well.
 */
STATICFNDEF void	gtmrecv_process_need_histinfo_msg(repl_needhistinfo_msg_ptr_t need_histinfo_msg, repl_histinfo *histinfo)
{
	boolean_t		maintain_rcvd_strm_histinfo, suppl_propagate_primary;
	int4			need_histinfo_num, need_histinfo_strm_num, status;
	recvpool_ctl_ptr_t	recvpool_ctl;
	seq_num			first_unprocessed_seqno, last_unprocessed_histinfo_seqno;
	seq_num			need_histinfo_seqno, last_valid_histinfo_seqno;

	assert(remote_side->endianness_known); /* ensure remote_side->cross_endian is reliable */
	if (!remote_side->cross_endian)
	{
		need_histinfo_seqno = need_histinfo_msg->seqno;
		need_histinfo_num = need_histinfo_msg->histinfo_num;
		need_histinfo_strm_num = need_histinfo_msg->strm_num;
	} else
	{
		need_histinfo_seqno = GTM_BYTESWAP_64(need_histinfo_msg->seqno);
		need_histinfo_num = GTM_BYTESWAP_32(need_histinfo_msg->histinfo_num);
		need_histinfo_strm_num = GTM_BYTESWAP_32(need_histinfo_msg->strm_num);
	}
	assert((INVALID_SUPPL_STRM == strm_index) || ((0 <= strm_index) && (MAX_SUPPL_STRMS > strm_index)));
	if (INVALID_SUPPL_STRM == strm_index)
	{	/* Both receiver and source sides are non-supplementary instances */
		if (REPL_PROTO_VER_SUPPLEMENTARY > remote_side->proto_ver)
		{	/* needhistinfo_msg.strm_num & histinfo_num are uninitialized in this case. Fix it */
			need_histinfo_strm_num = INVALID_SUPPL_STRM;
			need_histinfo_num = INVALID_HISTINFO_NUM;
		} else
		{
			assert(INVALID_SUPPL_STRM == need_histinfo_strm_num);
			assert(INVALID_HISTINFO_NUM == need_histinfo_num);
		}
		repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received REPL_NEED_HISTINFO message for seqno "INT8_FMT" "INT8_FMTX"\n",
			need_histinfo_seqno, need_histinfo_seqno);
	} else if (0 < strm_index)
	{	/* Receiver side is supplementary but Source side is a non-supplementary instance */
		assert(INVALID_SUPPL_STRM == need_histinfo_strm_num);
		assert(INVALID_HISTINFO_NUM == need_histinfo_num);
		repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received REPL_NEED_HISTINFO message for Stream %d : Seqno "
			INT8_FMT" "INT8_FMTX"\n", strm_index, need_histinfo_seqno, need_histinfo_seqno);
		need_histinfo_strm_num = strm_index;
	} else
	{	/* Both receiver and source sides are supplementary instances */
		/* strm_index is 0 at this point (already asserted above) */
		assert(INVALID_SUPPL_STRM != need_histinfo_strm_num);
		if ((INVALID_HISTINFO_NUM == need_histinfo_num) || (UNKNOWN_HISTINFO_NUM == need_histinfo_num))
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received REPL_NEED_HISTINFO message for Stream %d : "
				"Seqno "INT8_FMT" "INT8_FMTX"\n", need_histinfo_strm_num, need_histinfo_seqno, need_histinfo_seqno);
		else
		{
			assert(0 <= need_histinfo_num);
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received REPL_NEED_HISTINFO message for History Number %d\n",
				need_histinfo_num);
		}
	}
	/* The only two histinfo_num values that have special meaning are negative. So we can check for a valid value
	 * by checking for positive. Assert that below before doing the positive check.
	 */
	assert((0 > INVALID_HISTINFO_NUM) && (0 > UNKNOWN_HISTINFO_NUM));
	recvpool_ctl = recvpool.recvpool_ctl;
	if (0 <= need_histinfo_num)
	{	/* Handle simplest case first. Get the "need_histinfo_num"'th history record directly from the instance file */
		if (recvpool.gtmrecv_local->updateresync && (FD_INVALID != recvpool.gtmrecv_local->updresync_instfile_fd))
		{	/* The history record that is being requested needs to be found in the -updateresync input instance file
			 * (not the receiver side instance file).
			 */
			repl_log(gtmrecv_log_fp, TRUE, TRUE,
				"Searching for desired history in the -updateresync input instance file\n");
			gtmrecv_updresync_histinfo_get(need_histinfo_num, histinfo);
		} else
		{	/* The history record needs to be found in the receiver side instance file */
			repl_log(gtmrecv_log_fp, TRUE, TRUE,
				"Searching for the desired history in the replication instance file\n");
			grab_lock(jnlpool->jnlpool_dummy_reg, TRUE, GRAB_LOCK_ONLY);
			GTMRECV_ONLN_RLBK_CLNUP_IF_NEEDED;
				/* above macro will "return" if repl_connection_reset OR gtmrecv_wait_for_jnl_seqno is set */
			status = repl_inst_histinfo_get(need_histinfo_num, histinfo);
			rel_lock(jnlpool->jnlpool_dummy_reg);
			if (0 != status)
			{	/* Close the connection */
				assert(ERR_REPLINSTNOHIST == status);
				assert(FALSE);
				gtmrecv_autoshutdown();	/* should not return */
			}
		}
		maintain_rcvd_strm_histinfo = TRUE;
	} else if (UNKNOWN_HISTINFO_NUM == need_histinfo_num)
	{	/* This value was sent in a previous REPL_STRMINFO message for a particular non-zero stream # because
		 * we had not yet played this history record in the instance file. Return history record directly from
		 * where the receiver server saved a copy of the last unprocessed history record for this stream number.
		 */
		assert(need_histinfo_strm_num);
		assert(remote_side->is_supplementary);
		assert((0 < need_histinfo_strm_num) && (MAX_SUPPL_STRMS > need_histinfo_strm_num));
		repl_log(gtmrecv_log_fp, TRUE, TRUE, "Searching for desired history in the receive pool\n");
		*histinfo = recvpool_ctl->last_valid_strm_histinfo[need_histinfo_strm_num];
		maintain_rcvd_strm_histinfo = FALSE;
	} else
	{
		first_unprocessed_seqno = recvpool.upd_proc_local->read_jnl_seqno;
		repl_log(gtmrecv_log_fp, TRUE, FALSE, "Update process has processed upto seqno "INT8_FMT" "INT8_FMTX"\n",
			first_unprocessed_seqno, first_unprocessed_seqno);
		suppl_propagate_primary = remote_side->is_supplementary;
		if (!suppl_propagate_primary)
			last_valid_histinfo_seqno = recvpool_ctl->last_valid_histinfo.start_seqno;
		else
			last_valid_histinfo_seqno = recvpool_ctl->last_valid_strm_histinfo[0].start_seqno;
		repl_log(gtmrecv_log_fp, TRUE, TRUE,
			"Starting seqno of the last valid history in the receive pool is "INT8_FMT" "INT8_FMTX"\n",
			last_valid_histinfo_seqno, last_valid_histinfo_seqno);
		if (last_valid_histinfo_seqno >= first_unprocessed_seqno)
			last_unprocessed_histinfo_seqno = last_valid_histinfo_seqno;
		else
			last_unprocessed_histinfo_seqno = MAX_SEQNO;
		if (last_unprocessed_histinfo_seqno && (need_histinfo_seqno > last_unprocessed_histinfo_seqno))
		{
			/* NOTE0: The source server is requesting histinfo information for a seqno whose corresponding
			 *   histinfo has also not yet been processed by the update process (and hence not present in the
			 *   instance file). Find latest histinfo information that is stored in receive pool.
			 * NOTE1: Even though there could be more than one unprocessed history record in the receive pool,
			 *   the source should request only the last one for comparison. If the last history record on the
			 *   receiver matches on the source side too, replication can resume from there.  If not, the receiver
			 *   side should do a rollback (REPL_ROLLBACK_FIRST message). That is why it is enough to maintain
			 *   recvpool_ctl->last_valid_histinfo and not pointers to all the unprocessed histories.
			 * NOTE2: There is one exception to this and that is if the receiver server had already connected
			 *   to a source server and placed records in the receiver pool and then lost the connection and
			 *   reestablished connection (with the same or a different source server) AND had been started with
			 *   the -noresync option. In this case, it will not see a REPL_ROLLBACK_FIRST message but instead
			 *   will go back in history to find the first matching history. But in this case, the -noresync is
			 *   valid only for the FIRST connection and is cleared for future connections. So reconnections would
			 *   assume as if -noresync was not specified and hence will fall into the same REPL_ROLLBACK_FIRST
			 *   category as described in NOTE0. Assert it below.
			 * NOTE3: In the case of a propagating primary supplementary instance, we need to not just store the
			 *   last received history record but also one for each stream possible.
			 */
			assert(!recvpool.gtmrecv_local->updateresync);
			assert(!recvpool.gtmrecv_local->noresync);
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Searching for the desired history in the receive pool\n");
			if (!suppl_propagate_primary)
				memcpy(histinfo, &recvpool_ctl->last_valid_histinfo, SIZEOF(repl_histinfo));
			else
			{
				assert(!need_histinfo_strm_num);
				memcpy(histinfo, &recvpool_ctl->last_valid_strm_histinfo[0], SIZEOF(repl_histinfo));
			}
			assert(!recvpool_ctl->insert_strm_histinfo);
		} else if (recvpool.gtmrecv_local->updateresync && (FD_INVALID != recvpool.gtmrecv_local->updresync_instfile_fd))
		{	/* The receiver was started with -UPDATERESYNC=<INSTFILENAME>. The history record that is being requested
			 * needs to be found in the input instance file (not the receiver side instance file).
			 * Read the history record corresponding to need_histinfo_seqno.
			 */
			repl_log(gtmrecv_log_fp, TRUE, TRUE,
				"Searching for desired history in the -updateresync input instance file\n");
			gtmrecv_updresync_histinfo_find_seqno(need_histinfo_seqno, need_histinfo_strm_num, histinfo);
			assert(histinfo->start_seqno);
		} else
		{	/* The seqno has been processed by the update process. Hence the histinfo
			 * for this will be found in the instance file. Search there.
			 */
			assert(NULL != jnlpool->jnlpool_dummy_reg);
			repl_log(gtmrecv_log_fp, TRUE, TRUE,
				"Searching for the desired history in the replication instance file\n");
			grab_lock(jnlpool->jnlpool_dummy_reg, TRUE, GRAB_LOCK_ONLY);
			GTMRECV_ONLN_RLBK_CLNUP_IF_NEEDED;
			status = repl_inst_wrapper_histinfo_find_seqno(need_histinfo_seqno, need_histinfo_strm_num, histinfo);
			rel_lock(jnlpool->jnlpool_dummy_reg);
			if (0 != status)
			{	/* Close the connection */
				assert(ERR_REPLINSTNOHIST == status);
				gtmrecv_autoshutdown();	/* should not return */
			}
			assert((histinfo->histinfo_num != (jnlpool->repl_inst_filehdr->num_histinfo - 1))
				|| (histinfo->start_seqno == jnlpool->jnlpool_ctl->last_histinfo_seqno));
			if (0 < need_histinfo_strm_num)
			{	/* About to send to a non-supplementary instance. It does not understand strm_seqnos.
				 * So convert it back to a format it understands.
				 */
				CONVERT_SUPPL2NONSUPPL_HISTINFO(*histinfo);
			}
			assert(histinfo->start_seqno);
			assert(histinfo->start_seqno < need_histinfo_seqno);
		}
		maintain_rcvd_strm_histinfo = (0 == histinfo->strm_index);

	}
	if (maintain_rcvd_strm_histinfo && recvpool_ctl->insert_strm_histinfo)
	{
		assert(remote_side->is_supplementary);
		need_histinfo_strm_num = histinfo->strm_index;
		assert((0 <= need_histinfo_strm_num) && (MAX_SUPPL_STRMS > need_histinfo_strm_num));
		assert(IS_REPL_INST_UUID_NULL(recvpool_ctl->last_rcvd_strm_histinfo[need_histinfo_strm_num].lms_group));
		assert((FALSE == recvpool_ctl->is_valid_strm_histinfo[need_histinfo_strm_num])
			|| !memcmp(&recvpool_ctl->last_rcvd_strm_histinfo[need_histinfo_strm_num],
									histinfo, SIZEOF(repl_histinfo)));
		memcpy(&recvpool_ctl->last_rcvd_strm_histinfo[need_histinfo_strm_num], histinfo, SIZEOF(repl_histinfo));
		recvpool_ctl->is_valid_strm_histinfo[need_histinfo_strm_num] = TRUE;
		if (recvpool_ctl->max_strm_histinfo <= need_histinfo_strm_num)
			recvpool_ctl->max_strm_histinfo = need_histinfo_strm_num + 1;
	}
	return;
}

#ifdef GTM_TLS
/* The below logic is very similar to `gtmsource_exchange_tls_info' but is kept separate because of the following reasons:
 * (a) `gtmrecv_poll_actions' needs to be invoked as opposed to `gtmsource_poll_actions'.
 * (b) The arguments passed to `gtmrecv_poll_actions' are defined as static.
 * (c) The action taken after `gtmsource_poll_actions' and `gtmrecv_poll_actions' are different.
 *
 * The Receiver and Source servers exchanged TLS plugin library and 3rd party TLS library version information as part of
 * negotiating the startup of TLS. In the event of a handshake error, print that information as it may contribute to the
 * reason for the TLS handshake error. NOTE: Since gtmsource_exchange_tls_info() initiates the TLS negotiation, it has
 * the client TLS information. The Receiver side calls this function after it receives the same details and so those bits
 * of information must be passed to this function.
 */
STATICFNDEF boolean_t gtmrecv_exchange_tls_info(uint4 remote_API_ver, uint4 remote_lib_ver)
{
	int			errlen, poll_dir, status;
	static int		flags = GTMTLS_OP_FORCE_VERIFY_PEER | GTMTLS_OP_RENEGOTIATE_REQUESTED;
	char			*errp;
	boolean_t		pha_fallback;
	repl_tlsinfo_msg_t	reply;

	reply.type = REPL_TLS_INFO;
	reply.len = MIN_REPL_MSGLEN;
	reply.API_version = GTM_TLS_API_VERSION;
	reply.library_version = (uint4)tls_ctx->runtime_version;
	if (remote_side->cross_endian)
	{
		reply.API_version = GTM_BYTESWAP_32(reply.API_version);
		reply.library_version = GTM_BYTESWAP_32(reply.library_version);
	}
	gtmrecv_repl_send((repl_msg_ptr_t)&reply, REPL_TLS_INFO, SIZEOF(repl_tlsinfo_msg_t), "REPL_TLS_INFO", MAX_SEQNO);
	if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
		return FALSE;
	/* At this point, the both sides are ready for a TLS/SSL handshake. Create a TLS/SSL aware socket. */
	if (NULL == (repl_tls.sock = gtm_tls_socket(tls_ctx, repl_tls.sock, gtmrecv_sock_fd, repl_tls.id, flags)))
	{
		if (PLAINTEXT_FALLBACK)
			gtm_putmsg_csa(CSA_ARG(NULL) VARLSTCNT(6) MAKE_MSG_WARNING(ERR_TLSCONVSOCK), 0,
					ERR_TEXT, 2, LEN_AND_STR(gtm_tls_get_error(NULL)));
		else
		{	/* Close the connection and issue an error message */
			repl_close(&gtmrecv_sock_fd);
			repl_connection_reset = TRUE;
			gtm_putmsg_csa(CSA_ARG(NULL) VARLSTCNT(6) ERR_TLSCONVSOCK, 0,
					ERR_TEXT, 2, LEN_AND_STR(gtm_tls_get_error(NULL)));
			return FALSE;
		}
	} else
	{
		/* Do the actual handshake. */
		poll_dir = REPL_INVALID_POLL_DIRECTION;
		do
		{
			status = repl_do_tls_handshake(gtmrecv_log_fp, gtmrecv_sock_fd, TRUE, &poll_dir);
			assert(0 == data_len);
			gtmrecv_poll_actions(data_len, buff_unprocessed, buffp);
			if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
				return FALSE;
		} while ((GTMTLS_WANT_READ == status) || (GTMTLS_WANT_WRITE == status));
		if (SS_NORMAL == status) /* Where enabled, do post handshake auth */
			status = repl_do_tls_post_handshake(gtmrecv_log_fp, gtmrecv_sock_fd);
		if (SS_NORMAL == status)
		{
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Secure communication enabled using TLS/SSL protocol\n");
			if (GTMTLS_OP_PHA_EXT_NOT_RECEIVED & flags)	/* Re-enable PHA for subsequent connections */
				flags &= ~GTMTLS_OP_PHA_EXT_NOT_RECEIVED;
			return TRUE;
		} else if (REPL_CONN_RESET(status))
		{
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Attempt to connect() with TLS/SSL protocol failed. "
					"Status = %d; %s\n", status, STRERROR(status));
			repl_close(&gtmrecv_sock_fd);
			repl_connection_reset = TRUE;
			return FALSE;
		}
		errp = (0 > status) ? (char *)gtm_tls_get_error(repl_tls.sock) : STRERROR(status);
		if (OUT_BUFF_SIZE > (errlen = strlen(errp))) /* Append version information in one message */
			snprintf(errp + errlen, OUT_BUFF_SIZE - errlen, "; Local API:0x%08x, LIB:0x%08x; Remote API:0x%08x, LIB:0x%08x",
					GTM_TLS_API_VERSION, (uint4)tls_ctx->runtime_version, remote_API_ver, remote_lib_ver);
		pha_fallback = ((GTMTLS_OP_PHA_EXT_FALLBACK & repl_tls.sock->flags) && (-2 == status)
					&& !(GTMTLS_OP_PHA_EXT_NOT_RECEIVED & flags));
		gtm_putmsg_csa(CSA_ARG(NULL) VARLSTCNT(6) /* Plaintext/PHA fallback issue a WARNING */
				((PLAINTEXT_FALLBACK || pha_fallback) ? MAKE_MSG_WARNING(ERR_TLSHANDSHAKE) : ERR_TLSHANDSHAKE),
				0, ERR_TEXT, 2, LEN_AND_STR(errp));
		/* Handshake error. If PHA related, disable PHA and issue an error. Otherwise, follow plaintext fallback behavior */
		if ((!PLAINTEXT_FALLBACK) || ((GTMTLS_OP_PHA_EXT_FALLBACK & repl_tls.sock->flags) && (-2 == status)))
		{	/* Close the connection and issue an error message */
			if (pha_fallback)	/* PHA fallback enabled and extension not received; disable PHA */
			{
				repl_log(gtmrecv_log_fp, TRUE, TRUE, "Post-handshake fallback enabled. Disconnecting. "
						"Post Handshake Authentication temporarily disabled for the next TLS connection "
						"attempt.\n");
				flags |= GTMTLS_OP_PHA_EXT_NOT_RECEIVED;
			}
			repl_close(&gtmrecv_sock_fd);
			repl_connection_reset = TRUE;
			return FALSE;
		}
	}
	repl_log(gtmrecv_log_fp, TRUE, TRUE, "Plaintext fallback enabled. Closing and reconnecting without TLS/SSL.\n");
	repl_close(&gtmrecv_sock_fd);
	repl_connection_reset = TRUE;
	CLEAR_REPL_TLS_REQUESTED; /* As if -tlsid qualifier was never specified. */
	return FALSE;
}
#endif

STATICFNDEF void do_main_loop(boolean_t crash_restart)
{
	/* The work-horse of the Receiver Server */
	boolean_t			dont_reply_to_heartbeat = FALSE, is_repl_cmpc, pha_peer_cert_check = FALSE;
	boolean_t			uncmpfail, send_cross_endian, recvpool_prepared = FALSE, copied_to_recvpool = FALSE;
	gtmrecv_local_ptr_t		gtmrecv_local;
	gtm_time4_t			ack_time;
        char				print_msg[PROC_RECVOPS_PRINT_MSG_LEN];
        char				print_msg_t[PROC_RECVOPS_PRINT_MSG_LEN];
	int4				msghdrlen, strm_num, processed_hdrlen = 0;
	int4				need_histinfo_num;
	int				cmpret;
	int				msg_type = REPL_MSGTYPE_LAST, msg_len, hdr_msg_type, hdr_msg_len;
	int				torecv_len, recvd_len, recvd_this_iter;	/* needed for REPL_RECV_LOOP */
	int				tosend_len, sent_len, sent_this_iter;	/* needed for REPL_SEND_LOOP */
	int				status, poll_dir;			/* needed for REPL_{SEND,RECV}_LOOP */
	recvpool_ctl_ptr_t		recvpool_ctl;
	repl_cmpinfo_msg_ptr_t		cmptest_msg;
	repl_cmpinfo_msg_t		cmpsolve_msg;
	repl_cmpmsg_ptr_t		cmpmsgp;
	repl_heartbeat_msg_t		heartbeat;
	repl_histinfo			histinfo;
	repl_needhistinfo_msg_ptr_t	need_histinfo_msg;
	repl_needinst_msg_ptr_t		need_instinfo_msg;
	repl_needstrminfo_msg_ptr_t	need_strminfo_msg;
	repl_old_instinfo_msg_t		old_instinfo_msg;
	repl_old_needinst_msg_ptr_t	old_need_instinfo_msg;
	repl_start_msg_ptr_t		msgp;
	repl_start_reply_msg_t		*start_msg;
	seq_num				ack_seqno, temp_ack_seqno;
	seq_num				request_from, recvd_jnl_seqno;
	sgmnt_addrs			*repl_csa;
	uchar_ptr_t			old_buffp, buffp_start = NULL;
	uint4				recvd_start_flags, len;
	uLong				cmplen;
	uLongf				destlen;
	unsigned char			*msg_ptr;				/* needed for REPL_{SEND,RECV}_LOOP */
	unsigned char			remote_jnl_ver;
	upd_proc_local_ptr_t		upd_proc_local;
	repl_logfile_info_msg_t		*logfile_msgp, logfile_msg;
#	ifdef GTM_TLS
	repl_msg_t			renegotiate_msg;
	boolean_t			reneg_ack_sent;
	repl_tlsinfo_msg_t		*need_tlsinfo_msgp;
	uint4				remote_lib_ver, remote_API_ver;
#	endif
	DCL_THREADGBL_ACCESS;

	SETUP_THREADGBL_ACCESS;
	recvpool_ctl = recvpool.recvpool_ctl;
	upd_proc_local = recvpool.upd_proc_local;
	gtmrecv_local = recvpool.gtmrecv_local;
	gtmrecv_wait_for_jnl_seqno = FALSE;

	assert((NULL != jnlpool->jnlpool_dummy_reg) && jnlpool->jnlpool_dummy_reg->open);
	repl_csa = &FILE_INFO(jnlpool->jnlpool_dummy_reg)->s_addrs;
#	ifdef DEBUG
	assert(!repl_csa->hold_onto_crit);
	ASSERT_VALID_JNLPOOL(repl_csa);
#	endif
	/* If BAD_TRANS was written by the update process, it would have updated recvpool_ctl->jnl_seqno accordingly.
	 * Only otherwise, do we need to wait for it to write "recvpool_ctl->jnl_seqno".
	 */
	if (!gtmrecv_bad_trans_sent)
	{
		repl_log(gtmrecv_log_fp, TRUE, TRUE, "Waiting for Update Process to write jnl_seqno\n");
		while (QWEQ(recvpool_ctl->jnl_seqno, seq_num_zero))
		{
			SHORT_SLEEP(GTMRECV_WAIT_FOR_STARTJNLSEQNO);
			GTMRECV_POLL_ACTIONS(0, 0, NULL);
		}
		/* The call to "gtmrecv_poll_actions" above might have set the variable "gtmrecv_wait_for_jnl_seqno" to TRUE.
		 * In that case, we need to reset it to FALSE here as we are now going to wait for the jnl_seqno below.
		 * Not doing so will cause us to wait for jnl_seqno TWICE (once now and once when we later enter this function).
		 */
		gtmrecv_wait_for_jnl_seqno = FALSE;
		/* Since remote primary is multisite capable (otherwise we would have issued an error), we need to send the
		 * journal seqno of this instance for comparison. If the receiver has received more seqnos than have been
		 * processed by the update process, we should be sending the last received seqno across to avoid receiving
		 * duplicate and out-of-order seqnos. This is maintained in "recvpool_ctl->jnl_seqno" and is guaranteed to
		 * be greater than or equal to the journal seqno of this instance.
		 */
		request_from = recvpool_ctl->jnl_seqno;
		/* If this is the first time the update process initialized "recvpool_ctl->jnl_seqno", it should be
		 * equal to "jnlpool_ctl->jnl_seqno". But if the receiver had already connected and received a bunch
		 * of seqnos and if the update process did not process all of them and if the receiver disconnects
		 * and re-establishes the connection, the value of "recvpool_ctl->jnl_seqno" could be greater than
		 * "jnlpool_ctl->jnl_seqno" if there is non-zero backlog on the secondary. Assert accordingly.
		 * There is one exception to this and that is if this is a root primary supplementary instance.
		 * In that case, the receive pool talks about the non-supplementary instance stream jnl_seqnos whereas
		 * the jnlpool talks about the merged stream of jnl_seqnos (including any local updates). Therefore we
		 * cannot compare the two at all.
		 */
		assert((!jnlpool->jnlpool_ctl->upd_disabled && jnlpool->repl_inst_filehdr->is_supplementary)
			|| (recvpool_ctl->jnl_seqno >= jnlpool->jnlpool_ctl->jnl_seqno));
		assert(request_from);
		repl_log(gtmrecv_log_fp, TRUE, TRUE, "Requesting transactions from JNL_SEQNO "INT8_FMT" "INT8_FMTX"\n",
			request_from, request_from);
		/* Send (re)start JNL_SEQNO to Source Server.
		 * Note that even though we might know the endianness of the source side at this time, we still send this
		 * message in the native endian format. This keeps the logic on the source server side simple. This is the only
		 * exception to the general rule that the receiver server does all endian conversion for messages except the
		 * first one in the connection handshake. The source side knows to endian convert this (instead of expecting it
		 * to be in the source endian format) if this EPL_START_JNL_SEQNO message is not the first one in the connection.
		 * The only exception is if the source side is pre-V55000 AND we have already determined the endianness of the
		 * source side (i.e. the REPL_START_JNL_SEQNO message about to be sent is not the first one for this connection)
		 * AND the source is cross-endian. In that case, the pre-V55000 source does not know to handle a
		 * receiver-side-native-endian format message. So endian convert the message only in this case.
		 */
		repl_log(gtmrecv_log_fp, TRUE, TRUE, "Sending REPL_START_JNL_SEQNO message with seqno "INT8_FMT" "INT8_FMTX"\n",
			request_from, request_from);
		send_cross_endian = (remote_side->endianness_known && remote_side->cross_endian
						&& (REPL_PROTO_VER_SUPPLEMENTARY > remote_side->proto_ver));
		msgp = (repl_start_msg_ptr_t)gtmrecv_msgp;
		memset(msgp, 0, SIZEOF(*msgp));
		/* Since REPL_START_JNL_SEQNO is 0, there is no endian conversion necessary but for completeness we do it */
		msgp->type = send_cross_endian ? GTM_BYTESWAP_32(REPL_START_JNL_SEQNO) : REPL_START_JNL_SEQNO;
		if (send_cross_endian)
			*((seq_num *)msgp->start_seqno) = GTM_BYTESWAP_64(request_from);
		else
			*((seq_num *)msgp->start_seqno) = request_from;
		msgp->start_flags = START_FLAG_NONE;
		msgp->start_flags |= (gtmrecv_options.stopsourcefilter ? START_FLAG_STOPSRCFILTER : 0);
		/* If -UPDATERESYNC is specified then let pre-V55000 source server know so it does not ask for history record
		 * exchange. In case of post-V55000 source server, we expect a value to the -updateresync qualifier (the instance
		 * file name) and that is used for the history record exchange. In that case the source server ignores the
		 * START_FLAG_UPDATERESYNC bit and requests a history exchange anyways.
		 */
		msgp->start_flags |= (gtmrecv_local->updateresync ? START_FLAG_UPDATERESYNC : 0);
		/* Let source server know if -NORESYNC was specified in receiver server startup */
		msgp->start_flags |= (gtmrecv_local->noresync ? START_FLAG_NORESYNC : 0);
		msgp->start_flags |= START_FLAG_HASINFO;
		if (this_side->is_std_null_coll)
			msgp->start_flags |= START_FLAG_COLL_M;
		msgp->start_flags |= START_FLAG_VERSION_INFO;
		GTMTRIG_ONLY(msgp->start_flags |= START_FLAG_TRIGGER_SUPPORT;)
#		ifdef GTM_TLS
		if (REPL_TLS_REQUESTED)
		{
			assert(NULL != tls_ctx);		/* gtm_tls_init() must have already happened. */
			msgp->start_flags |= START_FLAG_ENABLE_TLS;
		}
#		endif
		if (send_cross_endian)
			msgp->start_flags = GTM_BYTESWAP_32(msgp->start_flags);
		msgp->jnl_ver = this_side->jnl_ver;
		msgp->proto_ver = REPL_PROTO_VER_THIS;
		msgp->node_endianness = NODE_ENDIANNESS;
		msgp->is_supplementary = jnlpool->repl_inst_filehdr->is_supplementary;
		msgp->len = send_cross_endian ? GTM_BYTESWAP_32(MIN_REPL_MSGLEN) : MIN_REPL_MSGLEN;
		msg_len = MIN_REPL_MSGLEN;
		REPL_SEND_LOOP(gtmrecv_sock_fd, msgp, msg_len, REPL_POLL_NOWAIT)
		{
			GTMRECV_POLL_ACTIONS(0, 0, NULL);
		}
		CHECK_REPL_SEND_LOOP_ERROR(status, "REPL_START_JNL_SEQNO");
	}
	gtmrecv_bad_trans_sent = FALSE;
	request_from = recvpool_ctl->jnl_seqno;
	assert(request_from >= seq_num_one);

	repl_log(gtmrecv_log_fp, TRUE, TRUE, "Waiting for REPL_WILL_RESTART_WITH_INFO or REPL_ROLLBACK_FIRST message\n");
	/* Receive journal data and place it in the Receive Pool */
	buff_start = (unsigned char *)gtmrecv_msgp;
	buffp = buff_start;
	buff_unprocessed = 0;
	data_len = 0;
	write_loc = recvpool_ctl->write;
	write_wrap = recvpool_ctl->write_wrap;

	repl_recv_data_recvd = 0;
	repl_recv_data_processed = 0;
	repl_recv_postfltr_data_procd = 0;
	repl_recv_lastlog_data_recvd = 0;
	repl_recv_lastlog_data_procd = 0;
	msghdrlen = REPL_MSG_HDRLEN;
	GTMTLS_ONLY(DEBUG_ONLY(reneg_ack_sent = FALSE));
	GTMTLS_ONLY(repl_tls.renegotiate_state = REPLTLS_RENEG_STATE_NONE);
	while (TRUE)
	{
		recvd_len = gtmrecv_max_repl_msglen - buff_unprocessed;
		GTMTLS_ONLY(poll_dir = REPL_INVALID_POLL_DIRECTION);
		while ((SS_NORMAL == (status = repl_recv(gtmrecv_sock_fd,
							(buffp + buff_unprocessed), &recvd_len, REPL_POLL_WAIT, &poll_dir)))
			       && (0 == recvd_len))
		{
			recvd_len = gtmrecv_max_repl_msglen - buff_unprocessed;
			if (xoff_sent)
			{
				DO_FLOW_CONTROL(write_loc);
			}
			if (xoff_sent && GTMRECV_XOFF_LOG_CNT <= xoff_msg_log_cnt)
			{	/* Update process is still running slow; force wait before logging any message. */
				SHORT_SLEEP(MILLISECS_IN_SEC - 1); /* Sleep for one second. */
				REPL_DPRINT1("Waiting for Update Process to clear recvpool space\n");
				xoff_msg_log_cnt = 0;
			} else if (xoff_sent)
				xoff_msg_log_cnt++;
			GTMRECV_POLL_ACTIONS(data_len, buff_unprocessed, buffp);
		}
		if (SS_NORMAL != status)
		{
			if (EREPL_RECV == repl_errno)
			{
				if (REPL_CONN_RESET(status))
				{
					repl_log(gtmrecv_log_fp, TRUE, TRUE, "Connection reset. Status = %d ; %s\n",
							status, STRERROR(status));
					repl_log_conn_info(gtmrecv_sock_fd, gtmrecv_log_fp, TRUE);
					SEND_SYSMSG_REPLCOMM(LEN_AND_LIT("Error in receiving from source. Error in recv"));
				}
			} else if (EREPL_SELECT == repl_errno)
			{
					repl_log(gtmrecv_log_fp, TRUE, TRUE, "Error in receiving from source."
						 " Error in select: %d ; %s\n", status, STRERROR(status));
					ISSUE_REPLCOMM_ERROR("Error in receiving from source. Error in select", status);
			}
			repl_connection_reset = TRUE;
			repl_close(&gtmrecv_sock_fd);
			return;
		}
		if (repl_connection_reset)
			return;
		if (pha_peer_cert_check && gtm_tls_did_post_hand_shake(repl_tls.sock))
		{
			repl_log(gtmrecv_log_fp, TRUE, TRUE, "Post-handshake authentication completed successfully\n");
			repl_log_tls_info(gtmrecv_log_fp, repl_tls.sock);
			pha_peer_cert_check = FALSE;
		}
#		ifdef REPL_CMP_SOLVE_TESTING
		/* Received communication from the source server, so we can cancel the timer */
		if (TREF(gtm_environment_init) && repl_cmp_solve_timer_set)
		{
			cancel_timer((TID)repl_cmp_solve_rcv_timeout);
			repl_cmp_solve_timer_set = FALSE;
		}
#		endif
		/* Something on the replication pipe - read it */
		REPL_DPRINT3("Pending data len : %d  Prev buff unprocessed : %ld\n", data_len, buff_unprocessed);
		buff_unprocessed += recvd_len;
		repl_recv_data_recvd += (qw_num)recvd_len;
		if (gtmrecv_logstats)
			repl_log(gtmrecv_log_fp, FALSE, FALSE, "Recvd : %d  Total : %ld\n", recvd_len, repl_recv_data_recvd);
		while (msghdrlen <= buff_unprocessed)
		{
			buffp_start = buffp;
			if (0 == data_len)
			{
				assert(0 == ((unsigned long)buffp % REPL_MSG_ALIGN));
				DEBUG_ONLY(recvpool_prepared = FALSE);
				if (!remote_side->endianness_known)
				{
					remote_side->endianness_known = TRUE;
					msg_type = ((repl_msg_ptr_t)buffp)->type;
					if (!((REPL_MSGTYPE_LAST > msg_type) || (REPL_MSGTYPE_LAST > GTM_BYTESWAP_32(msg_type))))
						WACKY_MESSAGE(msg_type, &gtmrecv_sock_fd, 1);	/* impossible message type */
					if ((REPL_MSGTYPE_LAST < msg_type) && (REPL_MSGTYPE_LAST > GTM_BYTESWAP_32(msg_type)))
					{
						remote_side->cross_endian = TRUE;
						repl_log(gtmrecv_log_fp, TRUE, TRUE, "Source and Receiver sides have opposite "
							"endianness\n");
					} else
					{
						remote_side->cross_endian = FALSE;
						repl_log(gtmrecv_log_fp, TRUE, TRUE, "Source and Receiver sides have same "
							"endianness\n");
					}
				}
				if (remote_side->cross_endian)
				{
					hdr_msg_type = GTM_BYTESWAP_32(((repl_msg_ptr_t)buffp)->type);
					hdr_msg_len = GTM_BYTESWAP_32(((repl_msg_ptr_t)buffp)->len);
				} else
				{
					hdr_msg_type = ((repl_msg_ptr_t)buffp)->type;
					hdr_msg_len = ((repl_msg_ptr_t)buffp)->len;
				}
				if (0 >= hdr_msg_len)
					WACKY_MESSAGE(hdr_msg_type, &gtmrecv_sock_fd, 2);	/* invalid length - overflow */
				msg_type = hdr_msg_type & REPL_TR_CMP_MSG_TYPE_MASK;
				if (REPL_TR_CMP_JNL_RECS == msg_type)
				{
					processed_hdrlen = REPL_MSG_HDRLEN;
					msg_len = hdr_msg_len - processed_hdrlen;
					gtmrecv_repl_cmpmsglen = msg_len;
					gtmrecv_repl_uncmpmsglen = (hdr_msg_type >> REPL_TR_CMP_MSG_TYPE_BITS);
					assert(0 < gtmrecv_repl_uncmpmsglen);
					assert(REPL_TR_CMP_THRESHOLD > gtmrecv_repl_uncmpmsglen);
					/* Since msg_len is compressed length, it need not be 8-byte aligned. Make it so
					 * since 8-byte aligned length would have been sent by the source server anyways.
					 */
					msg_len = ROUND_UP(msg_len, REPL_MSG_ALIGN);
					buffp += processed_hdrlen;
					exp_data_len = gtmrecv_repl_uncmpmsglen;
					buff_unprocessed -= processed_hdrlen;
					GTMRECV_SET_BUFF_TARGET_CMPBUFF(gtmrecv_repl_cmpmsglen, gtmrecv_repl_uncmpmsglen,
						gtmrecv_cur_cmpmsglen);
				} else if (REPL_TR_CMP_JNL_RECS2 == msg_type)
				{	/* A REPL_TR_CMP_JNL_RECS2 message is special in that it has a bigger message header.
					 * So check if unprocessed length is greater than the header. If not need to read more.
					 */
					msghdrlen = REPL_MSG_HDRLEN2;
					if (msghdrlen > buff_unprocessed)	/* Did not receive the full-header.            */
						break;				/* Break out of here and read more data first. */
					msghdrlen = REPL_MSG_HDRLEN; /* reset to regular msg hdr length for future messages */
					processed_hdrlen = REPL_MSG_HDRLEN2;
					cmpmsgp = (repl_cmpmsg_ptr_t)buffp;
					if (remote_side->cross_endian)
					{
						cmpmsgp->cmplen = GTM_BYTESWAP_32(cmpmsgp->cmplen);
						cmpmsgp->uncmplen = GTM_BYTESWAP_32(cmpmsgp->uncmplen);
					}
					gtmrecv_repl_cmpmsglen = cmpmsgp->cmplen;
					gtmrecv_repl_uncmpmsglen = cmpmsgp->uncmplen;
					assert(0 < gtmrecv_repl_uncmpmsglen);
					assert(REPL_TR_CMP_THRESHOLD <= gtmrecv_repl_uncmpmsglen);
					msg_len = hdr_msg_len - processed_hdrlen;
					/* Unlike REPL_TR_CMP_JNL_RECS message, msg_len is guaranteed to be 8-byte aligned here */
					buffp += processed_hdrlen;
					exp_data_len = gtmrecv_repl_uncmpmsglen;
					buff_unprocessed -= processed_hdrlen;
					GTMRECV_SET_BUFF_TARGET_CMPBUFF(gtmrecv_repl_cmpmsglen, gtmrecv_repl_uncmpmsglen,
						gtmrecv_cur_cmpmsglen);
				} else
				{
					processed_hdrlen = REPL_MSG_HDRLEN;
					msg_len = hdr_msg_len - processed_hdrlen;
					exp_data_len = msg_len;
					if (REPL_TR_JNL_RECS == msg_type || REPL_OLD_TRIPLE == msg_type || REPL_HISTREC == msg_type)
					{
						/* Target buffer is the receive pool. Prepare the receive pool for write (also
						 * checks if the transaction will fit in).
						 * Note: this is a special case where PREPARE_RECVPOOL_FOR_WRITE is not invoked
						 * right before COPY_TO_RECVPOOL because we want to prepare the receive pool just
						 * once even though the actual data (coming from the other end) might come in
						 * different pieces.
						 */
						PREPARE_RECVPOOL_FOR_WRITE(exp_data_len, 0);
						DEBUG_ONLY(recvpool_prepared = TRUE);
					} /* for REPL_TR_CMP_JNL_RECS{2}, receive pool is prepared after uncompression */
					buffp += processed_hdrlen;
					buff_unprocessed -= processed_hdrlen;
				}
				assert(0 <= buff_unprocessed);
				assert(0 == (msg_len % REPL_MSG_ALIGN));
				data_len = msg_len;
				assert(0 == (exp_data_len % REPL_MSG_ALIGN));
			}
			assert(0 == (data_len % REPL_MSG_ALIGN));
			buffered_data_len = ((data_len <= buff_unprocessed) ? data_len : buff_unprocessed);
			buffered_data_len = ROUND_DOWN2(buffered_data_len, REPL_MSG_ALIGN);
			old_buffp = buffp;
			buffp += buffered_data_len;
			buff_unprocessed -= buffered_data_len;
			assert(0 <= buff_unprocessed);
			data_len -= buffered_data_len;
			/* Once we have sent a REPL_RENEG_ACK, the only message we should get is the REPL_RENEG_COMPLETE. */
			assert(buffp > buff_start);
			assert(buffp <= (buff_start + gtmrecv_max_repl_msglen));
			assert(!reneg_ack_sent || (REPL_RENEG_COMPLETE == msg_type));
			copied_to_recvpool = FALSE;
			switch(msg_type)
			{
				case REPL_TR_JNL_RECS:
				case REPL_TR_CMP_JNL_RECS:
				case REPL_TR_CMP_JNL_RECS2:
				case REPL_OLD_TRIPLE:
				case REPL_HISTREC:
					is_repl_cmpc = ((REPL_TR_CMP_JNL_RECS == msg_type) || (REPL_TR_CMP_JNL_RECS2 == msg_type));
					if (!is_repl_cmpc)
					{
						assert(recvpool_prepared);
						COPY_TO_RECVPOOL(old_buffp, buffered_data_len);	/* uses and updates "write_loc" */
					} else
					{
						memcpy(gtmrecv_cmpmsgp + gtmrecv_cur_cmpmsglen, old_buffp, buffered_data_len);
						gtmrecv_cur_cmpmsglen += buffered_data_len;
						assert(gtmrecv_cur_cmpmsglen <= gtmrecv_max_repl_cmpmsglen);
					}
					/* The partial/complete message is copied either to the receive pool or the private
					 * compression buffer space. Set copied_to_recvpool to TRUE so that we can safely
					 * set buffp = buff_start if this is the last message in the current received buffer.
					 */
					copied_to_recvpool = TRUE;
					repl_recv_data_processed += (qw_num)buffered_data_len;
					if (0 == data_len)
					{
						process_tr_buff(msg_type);
						if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
							return;
					}
					break;

				case REPL_LOSTTNCOMPLETE:
					if (0 == data_len)
					{
						assert(REPL_PROTO_VER_MULTISITE <= remote_side->proto_ver);
						repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received REPL_LOSTTNCOMPLETE message\n");
						status = repl_inst_reset_zqgblmod_seqno_and_tn();
						assert(-1 == status || EXIT_ERR == status || SS_NORMAL == status);
						if (-1 == status)
						{	/* only reason we know currently for the above function to return -1 is due
							 * to a concurrent online rollback. In this case, we cannot continue
							 * and need to start afresh.
							 */
							gtmrecv_onln_rlbk_clnup();
							if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
								return;
						}
					}
					break;

				case REPL_HEARTBEAT:
					if (0 == data_len)
					{	/* Heartbeat msg contents start from buffp - msg_len */
						GTM_WHITE_BOX_TEST(WBTEST_REPL_HEARTBEAT_NO_ACK, dont_reply_to_heartbeat, TRUE);
						if (dont_reply_to_heartbeat)
						{
							dont_reply_to_heartbeat = FALSE;
							break;
						}
						memcpy(heartbeat.ack_seqno, buffp - msg_len, msg_len);
						assert(remote_side->endianness_known); /* only then is remote_side->cross_endian
											* reliable */
						if (!remote_side->cross_endian)
						{
							 ack_time = *(gtm_time4_t *)&heartbeat.ack_time[0];
							 memcpy((uchar_ptr_t)&ack_seqno,
								(uchar_ptr_t)&heartbeat.ack_seqno[0], SIZEOF(seq_num));
						} else
						{
							 ack_time = GTM_BYTESWAP_32(*(gtm_time4_t *)&heartbeat.ack_time[0]);
							 memcpy((uchar_ptr_t)&temp_ack_seqno,
								(uchar_ptr_t)&heartbeat.ack_seqno[0], SIZEOF(seq_num));
							 ack_seqno = GTM_BYTESWAP_64(temp_ack_seqno);
						}
						REPL_DPRINT4("HEARTBEAT received with time %ld SEQNO "INT8_FMT" at %ld\n",
							     ack_time, ack_seqno, time(NULL));
						ack_seqno = upd_proc_local->read_jnl_seqno;
						if (!remote_side->cross_endian)
						{
							heartbeat.type = REPL_HEARTBEAT;
							heartbeat.len = MIN_REPL_MSGLEN;
							memcpy((uchar_ptr_t)&heartbeat.ack_seqno[0],
								(uchar_ptr_t)&ack_seqno, SIZEOF(seq_num));

						} else
						{
							heartbeat.type = GTM_BYTESWAP_32(REPL_HEARTBEAT);
							heartbeat.len = GTM_BYTESWAP_32(MIN_REPL_MSGLEN);
							temp_ack_seqno = GTM_BYTESWAP_64(ack_seqno);
							memcpy((uchar_ptr_t)&heartbeat.ack_seqno[0],
								(uchar_ptr_t)&temp_ack_seqno, SIZEOF(seq_num));
						}
						REPL_SEND_LOOP(gtmrecv_sock_fd, &heartbeat, MIN_REPL_MSGLEN, REPL_POLL_NOWAIT)
						{
							GTMRECV_POLL_ACTIONS(data_len, buff_unprocessed, buffp);
						}
						CHECK_REPL_SEND_LOOP_ERROR(status, "REPL_HEARTBEAT");
						REPL_DPRINT4("HEARTBEAT sent with time %ld SEQNO "INT8_FMT" at %ld\n",
							     ack_time, ack_seqno, time(NULL));
					}
					break;

				case REPL_OLD_NEED_INSTANCE_INFO:
					if (0 == data_len)
					{
						assert(msg_len == MIN_REPL_MSGLEN - REPL_MSG_HDRLEN);
						old_need_instinfo_msg = (repl_old_needinst_msg_ptr_t)(buffp -
												msg_len - REPL_MSG_HDRLEN);
						repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received REPL_OLD_NEED_INSTANCE_INFO message"
							" from primary instance [%s]\n", old_need_instinfo_msg->instname);
						if (jnlpool->repl_inst_filehdr->is_supplementary)
						{	/* Issue REPL2OLD error because this is a supplementary instance and remote
							 * side runs a GT.M version that does not know the supplementary protocol */
							RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(6) ERR_REPL2OLD, 4,
								LEN_AND_STR(old_need_instinfo_msg->instname),
								LEN_AND_STR(jnlpool->repl_inst_filehdr->inst_info.this_instname));
						}
						/* The source server does not understand the supplementary protocol.
						 * So make sure -UPDATERESYNC if specified at receiver server startup
						 * had no value. If not, issue error.
						 */
						if (gtmrecv_local->updateresync
							&& (FD_INVALID != gtmrecv_local->updresync_instfile_fd))
						{
							RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(6) ERR_UPDSYNCINSTFILE, 0,
								ERR_TEXT, 2,
								LEN_AND_LIT("Source side is non-supplementary implies "
								"-UPDATERESYNC needs no value specified"));
						}
						/* Initialize the remote side protocol version from "proto_ver"
						 * field of this msg
						 */
						remote_side->proto_ver = old_need_instinfo_msg->proto_ver;
						assert(REPL_PROTO_VER_MULTISITE <= remote_side->proto_ver);
						assert(REPL_PROTO_VER_SUPPLEMENTARY > remote_side->proto_ver);
						/*************** Send REPL_OLD_INSTANCE_INFO message ***************/
						memset(&old_instinfo_msg, 0, SIZEOF(old_instinfo_msg));
						memcpy(old_instinfo_msg.instname,
							jnlpool->repl_inst_filehdr->inst_info.this_instname, MAX_INSTNAME_LEN - 1);
						grab_lock(jnlpool->jnlpool_dummy_reg, TRUE, GRAB_LOCK_ONLY);
						GTMRECV_ONLN_RLBK_CLNUP_IF_NEEDED;
						old_instinfo_msg.was_rootprimary = (unsigned char)repl_inst_was_rootprimary();
						rel_lock(jnlpool->jnlpool_dummy_reg);
						gtmrecv_repl_send((repl_msg_ptr_t)&old_instinfo_msg, REPL_OLD_INSTANCE_INFO,
								MIN_REPL_MSGLEN, "REPL_OLD_INSTANCE_INFO", MAX_SEQNO);
						if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
							return;
						/* Do not allow an instance which was formerly a root primary or which still
						 * has a non-zero value of "zqgblmod_seqno" to start up as a tertiary.
						 */
						if ((old_instinfo_msg.was_rootprimary || jnlpool->jnlpool_ctl->max_zqgblmod_seqno)
								&& !old_need_instinfo_msg->is_rootprimary)
						{
							gtm_putmsg_csa(CSA_ARG(NULL) VARLSTCNT(4) ERR_PRIMARYNOTROOT, 2,
								LEN_AND_STR((char *) old_need_instinfo_msg->instname));
							gtmrecv_autoshutdown();	/* should not return */
							assert(FALSE);
						}
						memcpy(jnlpool->jnlpool_ctl->primary_instname, old_need_instinfo_msg->instname,
							MAX_INSTNAME_LEN - 1);
					}
					break;

				case REPL_NEED_INSTINFO:
					if (0 == data_len)
					{
						assert(msg_len == SIZEOF(repl_needinst_msg_t) - REPL_MSG_HDRLEN);
						need_instinfo_msg = (repl_needinst_msg_ptr_t)(buffp - msg_len - REPL_MSG_HDRLEN);
						gtmrecv_check_and_send_instinfo(need_instinfo_msg, IS_RCVR_SRVR_TRUE);
						if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
							return;
					}
					break;

				case REPL_CMP_TEST:
					if (0 == data_len)
					{
						repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received REPL_CMP_TEST message\n");
						uncmpfail = FALSE;
						if (ZLIB_CMPLVL_NONE == gtm_zlib_cmp_level)
						{	/* Receiver does not have compression enabled in the first place.
							 * Send dummy REPL_CMP_SOLVE response message.
							 */
							repl_log(gtmrecv_log_fp, TRUE, TRUE, "Environment variable "
								"gtm_zlib_cmp_level specifies NO decompression (set to %d)\n",
								gtm_zlib_cmp_level);
							uncmpfail = TRUE;
						}
						assert(remote_side->endianness_known); /* ensure remote_side->cross_endian
											* is reliable */
						if (!uncmpfail)
						{
							assert(msg_len == REPL_MSG_CMPINFOLEN - REPL_MSG_HDRLEN);
							cmptest_msg = (repl_cmpinfo_msg_ptr_t)(buffp - msg_len - REPL_MSG_HDRLEN);
							if (!remote_side->cross_endian)
								cmplen = cmptest_msg->datalen;
							else
								cmplen = GTM_BYTESWAP_32(cmptest_msg->datalen);
							if (REPL_MSG_CMPEXPDATALEN < cmplen)
							{
								assert(FALSE);	/* since src srvr should not have sent such a msg */
								repl_log(gtmrecv_log_fp, TRUE, TRUE, "Compression test message "
									"has compressed data length (%d) greater than receiver "
									"allocated length (%d)\n", (int)cmplen,
									REPL_MSG_CMPEXPDATALEN);
								uncmpfail = TRUE;
							}
						}
						if (!uncmpfail)
						{
							destlen = REPL_MSG_CMPEXPDATALEN; /* initialize available
												     * decompressed buffer space */
							ZLIB_UNCOMPRESS(&cmpsolve_msg.data[0], destlen, &cmptest_msg->data[0],
										cmplen, cmpret);
							GTM_WHITE_BOX_TEST(WBTEST_REPL_TEST_UNCMP_ERROR, cmpret, Z_DATA_ERROR);
							switch(cmpret)
							{
								case Z_MEM_ERROR:
									assert(FALSE);
									repl_log(gtmrecv_log_fp, TRUE, TRUE,
										GTM_ZLIB_Z_MEM_ERROR_STR
										GTM_ZLIB_UNCMP_ERR_SOLVE_STR);
									break;
								case Z_BUF_ERROR:
									assert(FALSE);
									repl_log(gtmrecv_log_fp, TRUE, TRUE,
										GTM_ZLIB_Z_BUF_ERROR_STR
										GTM_ZLIB_UNCMP_ERR_SOLVE_STR);
									break;
								case Z_DATA_ERROR:
									assert(gtm_white_box_test_case_enabled
										&& (WBTEST_REPL_TEST_UNCMP_ERROR
											== gtm_white_box_test_case_number));
									repl_log(gtmrecv_log_fp, TRUE, TRUE,
										GTM_ZLIB_Z_DATA_ERROR_STR
										GTM_ZLIB_UNCMP_ERR_SOLVE_STR);
									break;
							}
							if (Z_OK != cmpret)
								uncmpfail = TRUE;
						}
						if (!uncmpfail)
						{
							cmpsolve_msg.datalen = (int4)destlen;
							GTM_WHITE_BOX_TEST(WBTEST_REPL_TEST_UNCMP_ERROR, cmpsolve_msg.datalen,
								REPL_MSG_CMPDATALEN - 1);
							if (REPL_MSG_CMPDATALEN != cmpsolve_msg.datalen)
							{	/* decompression did not yield precompressed data length */
								assert(gtm_white_box_test_case_enabled
									&& (WBTEST_REPL_TEST_UNCMP_ERROR
										== gtm_white_box_test_case_number));
								repl_log(gtmrecv_log_fp, TRUE, TRUE, GTM_ZLIB_UNCMPLEN_ERROR_STR
									"\n", cmpsolve_msg.datalen, REPL_MSG_CMPDATALEN);
								uncmpfail = TRUE;
							}
						}
						if (uncmpfail)
						{
							cmpsolve_msg.datalen = REPL_RCVR_CMP_TEST_FAIL;
							repl_log(gtmrecv_log_fp, TRUE, TRUE, GTM_ZLIB_UNCMPTRANSITION_STR);
						}
						if (remote_side->cross_endian)
							cmpsolve_msg.datalen = GTM_BYTESWAP_32(cmpsolve_msg.datalen);
						cmpsolve_msg.proto_ver = REPL_PROTO_VER_THIS;
#						ifdef REPL_CMP_SOLVE_TESTING
						if (TREF(gtm_environment_init))
						{
							start_timer((TID)repl_cmp_solve_rcv_timeout, 15 * 60 * 1000,
									repl_cmp_solve_rcv_timeout, 0, NULL);
							repl_cmp_solve_timer_set = TRUE;
						}
#						endif
						gtmrecv_repl_send((repl_msg_ptr_t)&cmpsolve_msg, REPL_CMP_SOLVE,
									REPL_MSG_CMPINFOLEN, "REPL_CMP_SOLVE", MAX_SEQNO);
						if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
							return;
						if (!uncmpfail)
							repl_zlib_cmp_level = gtm_zlib_cmp_level;
					}
					break;

				case REPL_NEED_STRMINFO:
					if (0 == data_len)
					{
						assert(REPL_PROTO_VER_SUPPLEMENTARY <= remote_side->proto_ver);
						assert(remote_side->is_supplementary);
						assert(msg_len == MIN_REPL_MSGLEN - REPL_MSG_HDRLEN);
						need_strminfo_msg = (repl_needstrminfo_msg_ptr_t)(buffp - msg_len
													- REPL_MSG_HDRLEN);
						gtmrecv_process_need_strminfo_msg(need_strminfo_msg);
						/* Check for error return from above function call */
						if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
							return;
					}
					break;

				case REPL_NEED_HISTINFO:
				/* case REPL_NEED_TRIPLE_INFO: too but that message has been renamed to REPL_NEED_HISTINFO */
					if (0 == data_len)
					{
						assert(REPL_PROTO_VER_MULTISITE <= remote_side->proto_ver);
						assert(msg_len == MIN_REPL_MSGLEN - REPL_MSG_HDRLEN);
						need_histinfo_msg = (repl_needhistinfo_msg_ptr_t)(buffp - msg_len
													- REPL_MSG_HDRLEN);
						gtmrecv_process_need_histinfo_msg(need_histinfo_msg, &histinfo);
						/* Check for error return from above function call */
						if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
							return;
						/* Send the histinfo */
						gtmrecv_send_histinfo(&histinfo);
						/* Check for error return from above function call as well */
						if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
							return;
					}
					break;

				case REPL_WILL_RESTART_WITH_INFO:
				case REPL_ROLLBACK_FIRST:
					if (0 != data_len)
						break;
					/* Have received a REPL_WILL_RESTART_WITH_INFO or REPL_ROLLBACK_FIRST message. If
					 * have not yet received a REPL_OLD_NEED_INSTANCE_INFO or REPL_NEED_INSTINFO message
					 * (which would have initialized "remote_side->proto_ver"), it
					 * means the remote side does not understand multi-site replication communication
					 * protocol. Note that down.
					 */
					if (REPL_PROTO_VER_UNINITIALIZED == remote_side->proto_ver)
					{	/*  Issue REPL2OLD error because primary is dual-site */
						RTS_ERROR_CSA_ABT(NULL, VARLSTCNT(6) ERR_REPL2OLD, 4,
							LEN_AND_STR(UNKNOWN_INSTNAME),
							LEN_AND_STR(jnlpool->repl_inst_filehdr->inst_info.this_instname));
					}
					/* Assert that endianness_known and cross_endian have already been initialized.
					 * This ensures that remote_side->cross_endian is reliable */
					assert(remote_side->endianness_known);
					assert(REPL_PROTO_VER_MULTISITE <= remote_side->proto_ver);
					assert(msg_len == MIN_REPL_MSGLEN - REPL_MSG_HDRLEN);
					start_msg = (repl_start_reply_msg_ptr_t)(buffp - msg_len - REPL_MSG_HDRLEN);
					assert((unsigned long)start_msg % SIZEOF(seq_num) == 0); /* alignment check */
					memcpy((uchar_ptr_t)&recvd_jnl_seqno,
						(uchar_ptr_t)start_msg->start_seqno, SIZEOF(seq_num));
					/* Assert that "node_endianness" field reflects our cross-endian understanding */
					BIGENDIAN_ONLY(assert((REPL_PROTO_VER_SUPPLEMENTARY > remote_side->proto_ver)
						|| (remote_side->cross_endian
							&& (LITTLE_ENDIAN_MARKER == start_msg->node_endianness))
						|| (!remote_side->cross_endian
							&& (BIG_ENDIAN_MARKER == start_msg->node_endianness))));
					LITTLEENDIAN_ONLY(assert((REPL_PROTO_VER_SUPPLEMENTARY > remote_side->proto_ver)
						|| (remote_side->cross_endian
							&& (BIG_ENDIAN_MARKER == start_msg->node_endianness))
						|| (!remote_side->cross_endian
							&& (LITTLE_ENDIAN_MARKER == start_msg->node_endianness))));
					if (remote_side->cross_endian)
						recvd_jnl_seqno = GTM_BYTESWAP_64(recvd_jnl_seqno);
					/* Handle REPL_ROLLBACK_FIRST case (easy one) first */
					if (REPL_ROLLBACK_FIRST == msg_type)
					{
						REPL_DPRINT1("Received REPL_ROLLBACK_FIRST message");
						sgtm_putmsg(print_msg, PROC_RECVOPS_PRINT_MSG_LEN, VARLSTCNT(4) ERR_REPLAHEAD, 2,
								LEN_AND_LIT(""));
						repl_log(gtmrecv_log_fp, TRUE, TRUE, print_msg);
						SNPRINTF(print_msg_t, SIZEOF(print_msg_t), "Replicating instance : "INT8_FMT" "
							"Originating instance : "INT8_FMT". ", request_from, recvd_jnl_seqno);
						sgtm_putmsg(print_msg, PROC_RECVOPS_PRINT_MSG_LEN, VARLSTCNT(4) ERR_TEXT, 2,
								 LEN_AND_STR(print_msg_t));
						repl_log(gtmrecv_log_fp, TRUE, TRUE, print_msg);

						if (gtmrecv_options.autorollback)
						{
							sgtm_putmsg(print_msg, PROC_RECVOPS_PRINT_MSG_LEN, VARLSTCNT(4) ERR_TEXT, 2,
								LEN_AND_LIT("Initiating automatic rollback to match the replicating"
								" instance with a prior in-sync originating instance state."));
							repl_log(gtmrecv_log_fp, TRUE, TRUE, print_msg);
							repl_connection_reset = TRUE;
							repl_close(&gtmrecv_sock_fd);
							repl_log(gtmrecv_log_fp, TRUE, TRUE, "Closing connection before starting "
								"ROLLBACK\n");
							if (SS_NORMAL != gtmrecv_start_onln_rlbk())
							{	/* gtmrecv_start_onln_rlbk() would have issued the appropriate
								 * error message.
								 */
								assert(FALSE);
								gtmrecv_autoshutdown(); /* should not return */
								assert(FALSE);
							}
							grab_lock(jnlpool->jnlpool_dummy_reg, TRUE, GRAB_LOCK_ONLY);
							GTMRECV_ONLN_RLBK_CLNUP_IF_NEEDED;
							/* The ONLINE ROLLBACK did not change the physical or the logical state.
							 * Otherwise the above macro would have returned to the caller. Since we
							 * have already disconnected the connection by now, we cannot resume the
							 * flow from this point on. Return to the calling function to continue
							 * reconnecting.
							 */
							rel_lock(jnlpool->jnlpool_dummy_reg);
							return;
						} else
						{
							sgtm_putmsg(print_msg, PROC_RECVOPS_PRINT_MSG_LEN, VARLSTCNT(4) ERR_TEXT, 2,
								LEN_AND_LIT(" The receiver server was not started with "
								"-AUTOROLLBACK. Perform MUPIP JOURNAL -ROLLBACK on the replicating"
								" instance to match the originating instance. Check the Source "
								"Server log to determine the last in-sync sequence number."));
							repl_log(gtmrecv_log_fp, TRUE, TRUE, print_msg);

						}
						gtmrecv_autoshutdown();	/* should not return */
						assert(FALSE);
						break;
					}
					/* Handle REPL_WILL_RESTART_WITH_INFO case now */
					if (jnlpool->repl_inst_filehdr->was_rootprimary && jnlpool->jnlpool_ctl->upd_disabled)
					{	/* This is the first time an instance that was formerly a root primary
						 * is brought up as an immediate secondary of the new root primary. Once
						 * fetchresync rollback has happened and the receiver and source server
						 * have communicated successfully, the instance file header field that
						 * indicates this was a root primary can be reset to FALSE as the zero
						 * or non-zeroness of the "zqgblmod_seqno" field in the respective
						 * database file headers henceforth controls whether this instance can
						 * be brought up as a tertiary or not. Flush changes to file on disk.
						 */
						grab_lock(jnlpool->jnlpool_dummy_reg, TRUE, GRAB_LOCK_ONLY);
						GTMRECV_ONLN_RLBK_CLNUP_IF_NEEDED;
						jnlpool->repl_inst_filehdr->was_rootprimary = FALSE;
						repl_inst_flush_filehdr();
						rel_lock(jnlpool->jnlpool_dummy_reg);
					}
					assert(REPL_WILL_RESTART_WITH_INFO == msg_type);
					repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received REPL_WILL_RESTART_WITH_INFO message"
							" with seqno "INT8_FMT" "INT8_FMTX"\n", recvd_jnl_seqno, recvd_jnl_seqno);
					remote_side->jnl_ver = start_msg->jnl_ver;
					remote_jnl_ver = remote_side->jnl_ver;
					REPL_DPRINT3("Local jnl ver is octal %o, remote jnl ver is octal %o\n",
						     this_side->jnl_ver, remote_jnl_ver);
					repl_check_jnlver_compat(!remote_side->cross_endian);
					/* older versions zero filler that was in place of start_msg->start_flags,
					 * so we are okay fetching start_msg->start_flags unconditionally.
					 */
					GET_ULONG(recvd_start_flags, start_msg->start_flags);
					if (remote_side->cross_endian)
						recvd_start_flags = GTM_BYTESWAP_32(recvd_start_flags);
					assert(remote_jnl_ver > V15_JNL_VER || 0 == recvd_start_flags);
					if (remote_jnl_ver <= V15_JNL_VER) /* safety in pro */
						recvd_start_flags = 0;
					remote_side->is_std_null_coll = (recvd_start_flags & START_FLAG_COLL_M) ?  TRUE : FALSE;
					if (remote_side->is_std_null_coll != this_side->is_std_null_coll)
						remote_side->null_subs_xform = (remote_side->is_std_null_coll
							?  STDNULL_TO_GTMNULL_COLL : GTMNULL_TO_STDNULL_COLL);
					else
						remote_side->null_subs_xform = FALSE;
						/* this sets null_subs_xform regardless of remote_jnl_ver */
					remote_side->is_supplementary = start_msg->is_supplementary;
					remote_side->trigger_supported = (recvd_start_flags & START_FLAG_TRIGGER_SUPPORT) ? TRUE
															  : FALSE;
#					ifdef GTM_TLS
					remote_side->tls_requested = (recvd_start_flags & START_FLAG_ENABLE_TLS) ? TRUE : FALSE;
					if (REPL_TLS_REQUESTED && !remote_side->tls_requested)
					{
						gtm_putmsg_csa(CSA_ARG(NULL) VARLSTCNT(6) ((PLAINTEXT_FALLBACK) ?
									 MAKE_MSG_WARNING(ERR_REPLNOTLS) : ERR_REPLNOTLS),
								4, LEN_AND_LIT(RCVR_SIDE_STR), LEN_AND_LIT(SRC_SIDE_STR));
						if (!PLAINTEXT_FALLBACK)
						{	/* Close the connection and return to waiting for a new one */
							repl_close(&gtmrecv_sock_fd);
							repl_connection_reset = TRUE;
							break;
						}
						repl_log(gtmrecv_log_fp, TRUE, TRUE, "Plaintext fallback enabled. "
								"Continuing without TLS/SSL.\n");
						CLEAR_REPL_TLS_REQUESTED; /* As if -tlsid qualifier was never specified. */
					}
					/* If we have not requested TLS/SSL and the originating side requested TLS/SSL, the latter
					 * reports the REPLNOTLS error and so we need not do anything. Hence, the below assert.
					 */
					assert(REPL_TLS_REQUESTED || !remote_side->tls_requested);
#					endif
					if (this_side->jnl_ver > remote_jnl_ver)
					{
						assert(JNL_VER_EARLIEST_REPL <= remote_jnl_ver);
						assert((remote_jnl_ver - JNL_VER_EARLIEST_REPL) < ARRAYSIZE(repl_filter_old2cur));
						assert((remote_jnl_ver - JNL_VER_EARLIEST_REPL) < ARRAYSIZE(repl_filter_cur2old));
						assert(IF_NONE != repl_filter_old2cur[remote_jnl_ver - JNL_VER_EARLIEST_REPL]);
						assert(IF_INVALID != repl_filter_old2cur[remote_jnl_ver - JNL_VER_EARLIEST_REPL]);
						/* reverse transformation should exist */
						assert(IF_INVALID != repl_filter_cur2old[remote_jnl_ver - JNL_VER_EARLIEST_REPL]);
						assert(IF_NONE != repl_filter_cur2old[remote_jnl_ver - JNL_VER_EARLIEST_REPL]);
						gtmrecv_filter |= INTERNAL_FILTER;
						gtmrecv_alloc_filter_buff(gtmrecv_max_repl_msglen);
					} else
					{
						gtmrecv_filter &= ~INTERNAL_FILTER;
						if (NO_FILTER == gtmrecv_filter)
							gtmrecv_free_filter_buff();
					}
					/* Don't send any more stopsourcefilter message */
					gtmrecv_options.stopsourcefilter = FALSE;
					assert(QWEQ(recvd_jnl_seqno, request_from)
							|| (jnlpool->repl_inst_filehdr->is_supplementary
								&& !jnlpool->jnlpool_ctl->upd_disabled && strm_index));
					assert(this_side->is_supplementary == jnlpool->repl_inst_filehdr->is_supplementary);
					if (this_side->is_supplementary && !remote_side->is_supplementary)
					{
						/* For the non-supplementary -> supplementary replication connection that happens
						 * for the very first time during the lifetime of this receiver server,
						 * (i.e. when upd_proc_local->read_jnl_seqno is 0), until this point in time,
						 * we cannot be sure what the starting point of the transmission is (partly due
						 * to -UPDATERESYNC but primarily due to -NORESYNC). Update process will be
						 * waiting for receiver to set read_jnl_seqno to a non-zero value. Finish it now.
						 * In case this is a reconnect, recvpool.upd_proc_local->read_jnl_seqno already
						 * reflects how much the update process has processed so that should be untouched
						 * by the receiver server as it will otherwise confuse the concurrently running
						 * update process (see <C9J02_003091_updproc_assert_fail_due_to_seqno_gap>).
						 * In addition do not touch recvpool_ctl->jnl_seqno in this case as we need to
						 * resume filling in the receive pool from where the previous connection stopped.
						 */
						assert(0 == GET_STRM_INDEX(recvd_jnl_seqno));
						if (0 == recvpool.upd_proc_local->read_jnl_seqno)
						{	/* Set read_jnl_seqno after jnl_seqno. Update process reads it in opposite
							 * order. Have memory barriers in between to ensure no out-of-order reads.
							 */
							recvpool.recvpool_ctl->jnl_seqno = recvd_jnl_seqno;
							SHM_WRITE_MEMORY_BARRIER;
							recvpool.upd_proc_local->read_jnl_seqno = recvd_jnl_seqno;
							repl_log(gtmrecv_log_fp, TRUE, TRUE, "Wrote upd_proc_local->read_jnl_seqno"
								" : "INT8_FMT" "INT8_FMTX"\n", recvd_jnl_seqno, recvd_jnl_seqno);
							/* If -NORESYNC was used in receiver startup, we don't want to use it
							 * anymore as future such connections could cause recvpool_ctl->jnl_seqno
							 * to be reset further backwards and will confuse the update process.
							 */
							if (recvpool.gtmrecv_local->noresync)
								recvpool.gtmrecv_local->noresync = FALSE;
						} else
						{
							assert(recvpool.recvpool_ctl->jnl_seqno
								>= recvpool.upd_proc_local->read_jnl_seqno);
							assert(!recvpool.gtmrecv_local->noresync);
						}
					}
					/* Now that recvpool.recvpool_ctl->jnl_seqno has been determined for sure (for
					 * non-supplementary or non-root-primary instances it is determined even before but
					 * otherwise it takes until now). Force a log on the first recv. This function uses
					 * recvpool.recvpool_ctl->jnl_seqno hence the placement here.
					 */
					gtmrecv_reinit_logseqno();
					break;

				case REPL_INST_NOHIST:
					if (0 == data_len)
					{
						assert(msg_len == MIN_REPL_MSGLEN - REPL_MSG_HDRLEN);
						repl_log(gtmrecv_log_fp, TRUE, TRUE, "Originating instance encountered a "
							"REPLINSTNOHIST error. JNL_SEQNO of this replicating instance precedes"
							" the current history in the originating instance file. "
							"Receiver server exiting.\n");
						gtmrecv_autoshutdown();	/* should not return */
						assert(FALSE);
					}
					break;

				case REPL_LOGFILE_INFO:
					if (0 == data_len)
					{
						logfile_msgp = (repl_logfile_info_msg_t *)(buffp - msg_len - REPL_MSG_HDRLEN);
						assert(REPL_PROTO_VER_REMOTE_LOGPATH <= logfile_msgp->proto_ver);
						if (remote_side->cross_endian)
						{
							logfile_msgp->fullpath_len = GTM_BYTESWAP_32(logfile_msgp->fullpath_len);
							logfile_msgp->pid = GTM_BYTESWAP_32(logfile_msgp->pid);
						}
						assert('\0' == logfile_msgp->fullpath[logfile_msgp->fullpath_len - 1]);
						repl_log(gtmrecv_log_fp, TRUE, TRUE, "Remote side source log file path is %s; "
								"Source Server PID = %d\n",
								logfile_msgp->fullpath, logfile_msgp->pid);
						/* Now, send our logfile path to the source side */
						assert(remote_side->endianness_known);
						len = repl_logfileinfo_get(recvpool.gtmrecv_local->log_file,
										&logfile_msg,
										remote_side->cross_endian,
										gtmrecv_log_fp);
						REPL_SEND_LOOP(gtmrecv_sock_fd, &logfile_msg, len, REPL_POLL_NOWAIT)
						{
							GTMRECV_POLL_ACTIONS(data_len, buff_unprocessed, buffp);
						}
						CHECK_REPL_SEND_LOOP_ERROR(status, "REPL_LOGFILE_INFO");
					}
					break;

#				ifdef GTM_TLS
				case REPL_NEED_TLS_INFO:
					if (0 != data_len)
						break;
					assert(REPL_PROTO_VER_TLS_SUPPORT <= REPL_PROTO_VER_THIS);
					assert(REPL_PROTO_VER_TLS_SUPPORT <= remote_side->proto_ver);
					assert(REPL_TLS_REQUESTED);
					assert(NULL != tls_ctx);
					assert(!repl_tls.enabled);
					need_tlsinfo_msgp = (repl_tlsinfo_msg_t *)(buffp - msg_len - REPL_MSG_HDRLEN);
					remote_API_ver = need_tlsinfo_msgp->API_version;
					remote_lib_ver = need_tlsinfo_msgp->library_version;
					if (remote_side->cross_endian)
					{
						remote_API_ver = GTM_BYTESWAP_32(remote_API_ver);
						remote_lib_ver = (uint4)GTM_BYTESWAP_32(remote_lib_ver);
					}
					repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received REPL_NEED_TLS_INFO message\n");
					repl_log(gtmrecv_log_fp, TRUE, TRUE, "  Remote side API version: 0x%08x\n", remote_API_ver);
					repl_log(gtmrecv_log_fp, TRUE, TRUE, "  Remote side Library version: 0x%08x\n",
														remote_lib_ver);
					if (!gtmrecv_exchange_tls_info(remote_API_ver, remote_lib_ver))
					{
						pha_peer_cert_check = FALSE;
						if (repl_connection_reset || gtmrecv_wait_for_jnl_seqno)
							return;
						assert(PLAINTEXT_FALLBACK);
						assert(!repl_tls.enabled);
					} else
					{	/* From here on, all communications are secured with SSL */
						repl_tls.enabled = TRUE;
						pha_peer_cert_check = gtm_tls_has_post_hand_shake(repl_tls.sock);
					}
					break;

				case REPL_RENEG_ACK_ME:
					if (0 != data_len)
						break;
					assert(!reneg_ack_sent);
					repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received %s acknowledgement request message\n",
							(gtm_tls_does_renegotiate(repl_tls.sock)) ? "renegotiation" : "key-update");
					gtmrecv_repl_send(&renegotiate_msg, REPL_RENEG_ACK, MIN_REPL_MSGLEN, "REPL_RENEG_ACK",
								MAX_SEQNO);
					DEBUG_ONLY(reneg_ack_sent = TRUE);
					repl_tls.renegotiate_state = REPLTLS_WAITING_FOR_RENEG_COMPLETE;
					break;

				case REPL_RENEG_COMPLETE:
					if (0 != data_len)
						break;
					assert(reneg_ack_sent);
					repl_log(gtmrecv_log_fp, TRUE, TRUE, "Received SSL/TLS connection %s complete message.\n",
							(gtm_tls_does_renegotiate(repl_tls.sock)) ? "renegotiation" : "key-update");
					DEBUG_ONLY(reneg_ack_sent = FALSE);
					if (gtm_tls_does_renegotiate(repl_tls.sock))	/* Announce results only if renegotiated */
						repl_log_tls_info(gtmrecv_log_fp, repl_tls.sock);
					repl_tls.renegotiate_state = REPLTLS_RENEG_STATE_NONE;
					break;
#				endif
				default:
					WACKY_MESSAGE(msg_type, &gtmrecv_sock_fd, 3);	/* undefined message type */
			}
			if (repl_connection_reset)
				return;
		}
		assert(0 == ((unsigned long)(buffp) % REPL_MSG_ALIGN));
		if (buff_start != buffp)
		{	/* We are at the tail of the current received buffer. */
			if ((0 != data_len) && !copied_to_recvpool)
			{	/* We have a complete header for a control message, but not the complete message. Move the message
				 * (including the header) to the beginning of the allocated buffer space. Restore the remaining
				 * length to buff_unprocessed and reset data_len so that the header is read again.
				 */
				assert(buffp_start);
				data_len = 0;
				buff_unprocessed += (buffered_data_len + processed_hdrlen);
				assert((0 <= buff_unprocessed) && (buff_unprocessed <= recvpool_size));
				if (buffp_start != buff_start)
					memmove(buff_start, buffp_start, buff_unprocessed);
			} else if (0 != buff_unprocessed)
			{	/* We have an incomplete header. Move it to the beginning of the allocated buffer space. */
				memmove(buff_start, buffp, buff_unprocessed);
			}
			buffp = buff_start;
		}
		GTMRECV_POLL_ACTIONS(data_len, buff_unprocessed, buffp);
	}
}

void repl_cmp_solve_rcv_timeout(void)
{
	assertpro(FALSE);
}

STATICFNDEF void gtmrecv_heartbeat_timer(TID tid, int4 interval_len, int *interval_ptr)
{
	assert(0 != gtmrecv_now);
	UNIX_ONLY(assert(*interval_ptr == heartbeat_period);)	/* interval_len and interval_ptr are dummies on VMS */
	gtmrecv_now += heartbeat_period;
	REPL_DPRINT2("Starting heartbeat timer with %d s\n", heartbeat_period);
	start_timer((TID)gtmrecv_heartbeat_timer, heartbeat_period * 1000, gtmrecv_heartbeat_timer, SIZEOF(heartbeat_period),
			&heartbeat_period); /* start_timer expects time interval in milli seconds, heartbeat_period is in seconds */
}

STATICFNDEF void gtmrecv_main_loop(boolean_t crash_restart)
{
	assert(FD_INVALID == gtmrecv_sock_fd);
	gtmrecv_poll_actions(0, 0, NULL); /* Clear any pending bad trans */
	gtmrecv_est_conn();
	gtmrecv_bad_trans_sent = FALSE; /* this assignment should be after gtmrecv_est_conn since gtmrecv_est_conn can
					 * potentially call gtmrecv_poll_actions. If the timing is right,
					 * gtmrecv_poll_actions might set this variable to TRUE if the update process sets
					 * bad_trans in the recvpool. When we are (re)establishing connection with the
					 * source server, there is no point in doing bad trans processing. Also, we have
					 * to send START_JNL_SEQNO message to the source server. If not, there will be a
					 * deadlock with the source and receiver servers waiting for each other to send
					 * a message. */
	repl_recv_prev_log_time = gtmrecv_now;
	while (!repl_connection_reset)
		do_main_loop(crash_restart);
	return;
}

void gtmrecv_process(boolean_t crash_restart)
{
	recvpool_ctl_ptr_t	recvpool_ctl;
	upd_proc_local_ptr_t	upd_proc_local;
	gtmrecv_local_ptr_t	gtmrecv_local;

	if (ZLIB_CMPLVL_NONE != gtm_zlib_cmp_level)
		gtm_zlib_init();	/* Open zlib shared library for compression/decompression */
#	ifdef GTM_TLS
	if (REPL_TLS_REQUESTED)
	{
		repl_do_tls_init(gtmrecv_log_fp);
		assert(REPL_TLS_REQUESTED || PLAINTEXT_FALLBACK);
	}
#	endif
	recvpool_ctl = recvpool.recvpool_ctl;
	upd_proc_local = recvpool.upd_proc_local;
	gtmrecv_local = recvpool.gtmrecv_local;

	/* Check all message sizes are the same size (32 bytes = MIN_REPL_MSGLEN) except for the REPL_OLD_TRIPLE message
	 * (repl_histrec_msg_t structure) which is 8 bytes more. Pre-supplementary, the receiver server knew to handle
	 * different sized messages only for a few messages types REPL_TR_JNL_RECS, REPL_OLD_TRIPLE and REPL_CMP_SOLVE.
	 * But post-supplementary it knows to handle different sized messages for various additional message types
	 * (including REPL_NEED_INSTINFO, REPL_INSTINFO, REPL_HISTREC).
	 */
	assert(MIN_REPL_MSGLEN == SIZEOF(repl_start_msg_t));
	assert(MIN_REPL_MSGLEN == SIZEOF(repl_start_reply_msg_t));
	assert(MIN_REPL_MSGLEN == SIZEOF(repl_resync_msg_t));
	assert(MIN_REPL_MSGLEN == SIZEOF(repl_old_needinst_msg_t));
	assert(MIN_REPL_MSGLEN <  SIZEOF(repl_needinst_msg_t));
	assert(MIN_REPL_MSGLEN == SIZEOF(repl_needhistinfo_msg_t));
	assert(MIN_REPL_MSGLEN == SIZEOF(repl_old_instinfo_msg_t));
	assert(MIN_REPL_MSGLEN <  SIZEOF(repl_instinfo_msg_t));
	assert(MIN_REPL_MSGLEN == SIZEOF(repl_histinfo1_msg_t));
	assert(MIN_REPL_MSGLEN == SIZEOF(repl_histinfo2_msg_t));
	assert(MIN_REPL_MSGLEN < SIZEOF(repl_histinfo_msg_t));
	assert(MIN_REPL_MSGLEN < SIZEOF(repl_old_triple_msg_t));
	assert(MIN_REPL_MSGLEN < SIZEOF(repl_histrec_msg_t));
	assert(MIN_REPL_MSGLEN == SIZEOF(repl_heartbeat_msg_t));
	assert(REPL_POLL_WAIT < MILLISECS_IN_SEC);
	recvpool_size = recvpool_ctl->recvpool_size;
	recvpool_high_watermark = (long)((float)RECVPOOL_HIGH_WATERMARK_PCTG / 100 * recvpool_size);
	recvpool_low_watermark  = (long)((float)RECVPOOL_LOW_WATERMARK_PCTG  / 100 * recvpool_size);
	if ((long)((float)(RECVPOOL_HIGH_WATERMARK_PCTG - RECVPOOL_LOW_WATERMARK_PCTG) / 100 * recvpool_size) >=
			RECVPOOL_XON_TRIGGER_SIZE)
	{ /* for large receive pools, the difference between high and low watermarks as computed above may be too large that
	   * we may not send XON quickly enough. Limit the difference to RECVPOOL_XON_TRIGGER_SIZE */
		recvpool_low_watermark = recvpool_high_watermark - RECVPOOL_XON_TRIGGER_SIZE;
	}
	REPL_DPRINT4("RECVPOOL HIGH WATERMARK is %ld, LOW WATERMARK is %ld, Receive pool size is %ld\n",
			recvpool_high_watermark, recvpool_low_watermark, recvpool_size);
	gtmrecv_alloc_msgbuff();
	gtmrecv_now = time(NULL);
	heartbeat_period = GTMRECV_HEARTBEAT_PERIOD; /* time keeper, well sorta */
	start_timer((TID)gtmrecv_heartbeat_timer, heartbeat_period * 1000, gtmrecv_heartbeat_timer, SIZEOF(heartbeat_period),
			&heartbeat_period); /* start_timer expects time interval in milli seconds, heartbeat_period is in seconds */
	do
	{
		gtmrecv_main_loop(crash_restart);
	} while (repl_connection_reset);
	assertpro(FALSE); /* shouldn't reach here */
	return;
}