Postgres-xl GTM（全局事务管理器 Globale Transaction Manager）GTM Master & Standby 子线程

川长思鸟来 2022-10-05 01:58 99阅读 0赞

## GTM Master ##

GTM Master在main（main.c文件）中按以下顺序初始化。它非常简单：设置配置、初始化主线程、从文件恢复信息和接受连接  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70]  
ServerLoop()来自postmaster.c的同名函数。它等待新的连接，如果需要，连接到备用，然后调用GTMAddConnection()来创建一个线程来处理连接。如果它检测到要中止的信号，它会将信息保存到控制文件并退出。  
GTM\_ThreadMain()是一个带有消息循环的主函数。此函数似乎是从postmaster.c:BackendRun()复制的。首先它初始化了很多东西；例如，内存上下文、连接、消息缓冲区和异常堆栈。与PostgreSQL一样，子函数中的一个信号和一个异常是通过long jump来实现的，因此使用sigsetjmp()注册一个异常处理程序。

### 数据结构 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center]

### 客户端连接 ###

ServerLoop函数涉及GTM Master新建客户端连接并为新建连接创建GTM Threads子进程部分的代码。ConnCreate函数创建Port数据结构，gtm\_standby\_connect\_to\_standby创建GTM\_Conn数据结构。

Port	   *port;
    port = ConnCreate(ListenSocket[i]);
    if (port){ 
    	GTM_Conn *standby = NULL;
    	standby = gtm_standby_connect_to_standby();
    	if (GTMAddConnection(port, standby) != STATUS_OK){ 
    		gtm_standby_disconnect_from_standby(standby);
    		StreamClose(port->sock);
    		ConnFree(port);

GTMAddConnection函数将Port数据结构和GTM\_Conn数据结构连接到GTM\_ConnectionInfo数据结构相应成员中，然后调用GTM\_ThreadCreate函数。

static int GTMAddConnection(Port *port, GTM_Conn *standby) { 
    	GTM_ConnectionInfo *conninfo = NULL;
    	conninfo = (GTM_ConnectionInfo *)palloc0(sizeof (GTM_ConnectionInfo));
    	elog(DEBUG3, "Started new connection");
    	conninfo->con_port = port;
    	/* Add a connection to the standby. */
    	if (standby != NULL)
    		conninfo->standby = standby;
    	/* XXX Start the thread */
    	if (GTM_ThreadCreate(conninfo, GTM_ThreadMain) == NULL)
    		return STATUS_ERROR;
    	return STATUS_OK;
    }

GTM\_ThreadCreate函数负责为线程设置各种内存上下文，以及向**线程管理器**注册该线程。创建成功后，线程将开始运行给定的“start routine”， 线程信息返回给调用进程，函数步骤如下：

1.  在内存上下文之外分配。我们还在线程特定的信息中跟踪各种线程持有的锁。 内存管理器也使用一个这样的锁thrinfo->thr\_lock，它在 thrinfo 被释放时被获取/释放。 我们不希望锁定信息在该过程中丢失。线程清理例程应确保释放此分配，避免任何内存泄漏。
2.  thrinfo->thr\_status = GTM\_THREAD\_STARTING 进程状态在这里被设置为GTM\_THREAD\_STARTING，由进程运行时改变该标志。
3.  向全局数组中注册ThreadInfo结构体。
4.  在实际启动线程之前设置内存contextes，TopThreadContext是TopMemoryContext的子级，它将在主进程或此线程的生命周期中持续存在。其他线程之间不共享线程上下文。
5.  由于线程没有启动，TopMemoryContext仍然指向调用线程的上下文。
6.  每个线程都有自己的ErrorContext和主进程的ErrorContext的子线程。这是特定于线程的上下文，不能在其他线程之间共享。
7.  将sn\_xip指针设置为指向静态分配的数组。
8.  现在开始创建线程。线程将开始执行给定的“startroutine”。thrinfo结构也传递给线程。任何附加参数都应通过thrinfo结构传递。并将thrinfo结构返回给调用者。
9.  确保在线程退出时释放资源(我们以前是在GTM\_ThreadMainWrapper内部执行的，但是在GTM\_ThreadMainWrapper开始执行时thrinfo->thr\_id可能没有设置，这可能是在无效的thr\_id上调用函数。

GTM_ThreadInfo *GTM_ThreadCreate(GTM_ConnectionInfo *conninfo, void *(* startroutine)(void *)) { 
    	GTM_ThreadInfo *thrinfo;
    	int err;
    	thrinfo = (GTM_ThreadInfo *)malloc(sizeof (GTM_ThreadInfo));
    	memset(thrinfo, 0, sizeof (GTM_ThreadInfo));
    	thrinfo->thr_conn = conninfo;
    	GTM_RWLockInit(&thrinfo->thr_lock);
    	thrinfo->thr_status = GTM_THREAD_STARTING;
    	if (GTM_ThreadAdd(thrinfo) == -1){ 
    		GTM_RWLockDestroy(&thrinfo->thr_lock);
    		free(thrinfo);
    		return NULL;
    	}
    	thrinfo->thr_thread_context = AllocSetContextCreate(TopMemoryContext,"TopMemoryContext",ALLOCSET_DEFAULT_MINSIZE,ALLOCSET_DEFAULT_INITSIZE,ALLOCSET_DEFAULT_MAXSIZE,false);
    	thrinfo->thr_parent_context = TopMemoryContext;
    	thrinfo->thr_error_context = AllocSetContextCreate(ErrorContext,"ErrorContext",8 * 1024,8 * 1024,8 * 1024,false);
    	thrinfo->thr_snapshot.sn_xip = thrinfo->thr_xip;
    	thrinfo->thr_startroutine = startroutine;
    	if ((err = pthread_create(&thrinfo->thr_id, NULL, GTM_ThreadMainWrapper,thrinfo))){ 
    		ereport(LOG,(err,errmsg("Failed to create a new thread: error %s", strerror(err))));
    		MemoryContextDelete(thrinfo->thr_error_context);
    		MemoryContextDelete(thrinfo->thr_thread_context);
    		GTM_RWLockDestroy(&thrinfo->thr_lock);
    		GTM_ThreadRemove(thrinfo);
    		return NULL;
    	}
    	pthread_detach(thrinfo->thr_id);
    	return thrinfo;
    }

### GTM Threads ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 1]

### 如何创建 ###

GTM Threads使用pthread\_create函数创建，其执行的函数是GTM\_ThreadMainWrapper，参数为thrinfo，而GTM\_ThreadMainWrapper会在进入时初始化，设置退出时cleanup handler，然后调用thrinfo->thr\_startroutine，也就是执行GTM\_ThreadCreate的函数指针参数startroutine。

GTM_ThreadInfo *GTM_ThreadCreate(GTM_ConnectionInfo *conninfo, void *(* startroutine)(void *)) { 
        ...
        thrinfo->thr_startroutine = startroutine;
        if ((err = pthread_create(&thrinfo->thr_id, NULL, GTM_ThreadMainWrapper,thrinfo))){ 
        ...
    }
    void *GTM_ThreadMainWrapper(void *argp) { 
    	GTM_ThreadInfo *thrinfo = (GTM_ThreadInfo *)argp;
    	SetMyThreadInfo(thrinfo);
    	MemoryContextSwitchTo(TopMemoryContext);
    	pthread_cleanup_push(GTM_ThreadCleanup, thrinfo);
    	thrinfo->thr_startroutine(thrinfo);
    	pthread_cleanup_pop(1);
    	return thrinfo;
    }

### 关键结构体成员 ###

子线程的内存上下文：

1.  TopThreadContext是TopMemoryContext的子级，它将在主进程或此线程的生命周期中持续存在。其他子线程之间不共享线程上下文。
2.  每个线程都有自己的ErrorContext，是主进程的ErrorContext的子线程。这是特定于线程的上下文，不能在其他线程之间共享。

thrinfo->thr_parent_context = TopMemoryContext;
    thrinfo->thr_thread_context = AllocSetContextCreate(TopMemoryContext,"TopMemoryContext",ALLOCSET_DEFAULT_MINSIZE,ALLOCSET_DEFAULT_INITSIZE,ALLOCSET_DEFAULT_MAXSIZE,false);
    thrinfo->thr_error_context = AllocSetContextCreate(ErrorContext,"ErrorContext",8 * 1024,8 * 1024,8 * 1024,false);
    thrinfo->thr_message_context; 
    thrinfo->thr_current_context;

子进程的内部锁：`GTM_RWLock thr_lock;`

该线程持有的锁的列表：`GTM_RWLock *thr_rwlocks_held[GTM_MAX_SIMUL_RWLOCKS];`

该线程持有的互斥锁的列表：`GTM_MutexLock *thr_mutexlocks_held[GTM_MAX_SIMUL_MUTEX];`

子线程运行状态标志：`GTM_ThreadStatus thr_status;`

typedef enum GTM_ThreadStatus { 
     GTM_THREAD_STARTING,
     GTM_THREAD_RUNNING,
     GTM_THREAD_EXITING,
     GTM_THREAD_BACKUP,   
     GTM_THREAD_INVALID
    } GTM_ThreadStatus;

### gtm\_thread.c提供的函数 ###

`GTM_ThreadInfo *GTM_ThreadCreate(GTM_ConnectionInfo *conninfo, void *(* startroutine)(void *))` 创建新进程并将给定连接赋给进程  
`static void GTM_ThreadCleanup(void *argp)` 进程清理函数  
`int GTM_ThreadAdd(GTM_ThreadInfo *thrinfo)` 将给定thrinfo结构体加入全局数组中，如果全局数组不够就扩展  
`int GTM_ThreadRemove(GTM_ThreadInfo *thrinfo)` 从全局数组中删除相应的thrinfo结构体  
`void * GTM_ThreadMainWrapper(void *argp)` 进程执行函数的包装函数  
`void GTM_DoForAllOtherThreads(void (* process_routine)(GTM_ThreadInfo *))` 为其他GTM子进程调用process\_routine函数  
这些函数依赖的C库函数如下所示：

//pthread_create是类Unix操作系统（Unix、Linux、Mac OS X等）的创建线程的函数。它的功能是创建线程（实际上就是确定调用该线程函数的入口点），在线程创建以后，就开始运行相关的线程函数。
    int pthread_create(pthread_t *tidp,const pthread_attr_t *attr, void *(*start_rtn)(void*),void *arg);
    //pthread_cleanup_push来注册清理函数rtn,这个函数有一个参数arg。在以下三种情形之一发生时，注册的清理函数被执行：
    //　　1）调用pthread_exit。
    //　　2）作为对取消线程请求(pthread_cancel)的响应。
    //　　3）以非0参数调用pthread_cleanup_pop。
    void pthread_cleanup_push(void (*rtn)(void *)，void *arg);
    void pthread_cleanup_pop(int execute);
    //使用pthread_setspecific可以为指定线程特定数据键设置线程特定绑定
    //形参：
    // key：需要关联的键
    // value：指向需要关联的数据
    int pthread_setspecific(pthread_key_t key, const void *value);
    //使用pthread_getspecific获取调用线程的键绑定，并将该绑定存储在value指向的位置中
    //形参：
    // key：需要获取数据的键
    void *pthread_getspecific(pthread_key_t key);

### gtm\_lock.c提供的函数 ###

`bool GTM_RWLockAcquire(GTM_RWLock *lock, GTM_LockMode mode)` 获取指定模式(write、read、)的锁  
`bool GTM_RWLockRelease(GTM_RWLock *lock)` 释放锁  
`int GTM_RWLockInit(GTM_RWLock *lock)` 初始化锁  
`int GTM_RWLockDestroy(GTM_RWLock *lock)` 销毁锁  
`void GTM_RWLockReleaseAll(void)` 释放线程持有的所有锁

`int GTM_MutexLockInit(GTM_MutexLock *lock)` 初始化互斥锁  
`int GTM_MutexLockDestroy(GTM_MutexLock *lock)` 销毁互斥锁  
`bool GTM_MutexLockAcquire(GTM_MutexLock *lock)` 获取互斥锁  
`bool GTM_MutexLockRelease(GTM_MutexLock *lock)` 释放互斥锁  
`void GTM_MutexLockReleaseAll(void)` 释放线程持有的所有互斥锁

`int GTM_CVInit(GTM_CV *cv)` 初始化条件变量  
`int GTM_CVBcast(GTM_CV *cv)` 唤醒所有等待条件变量的线程  
`int GTM_CVSignal(GTM_CV *cv)` 唤醒一个等待条件变量的线程  
`int GTM_CVWait(GTM_CV *cv, GTM_MutexLock *lock)` 等待一个条件变量，调用者必须已经获得互斥锁

这些函数依赖的C库函数如下所示：

pthread_rwlock_wrlock
    pthread_rwlock_rdlock
    pthread_rwlock_unlock
    pthread_rwlock_init
    pthread_rwlock_destroy
    pthread_mutex_init
    pthread_mutex_destroy
    pthread_mutex_lock
    pthread_mutex_unlock
    pthread_cond_init
    pthread_cond_broadcast
    pthread_cond_signal
    pthread_cond_wait

### GTM\_ThreadMain子线程真正运行的函数 ###

GTM\_ThreadMain是子线程真正运行的函数，其主要步骤如下：

1.  创建内存上下文，main loop每次迭代都会重置MessageContext,比如每次处理完客户端发来的命令信息
2.  获取线程局部锁，防止来自GTM Standby的连接更新GTM-Standby注册信息
3.  这里希望收到启动信息，该信息类型为REGISTER\_COORD（协调节点注册），后面紧跟4字节协调节点ID
4.  如果客户端之前连接过，就用之前GTM Master为该客户端分配的identifier
5.  向客户端发送dummy认证请求信息R，信息中包含客户identifier
6.  设置local sigjmp\_buf，主循环开始
7.  释放Message内存上下文中所有存储，在清空的消息内存上下文中创建新query输入缓冲
8.  释放线程局部锁，GTM-Standby注册信息可以在读命令操作期间更新
9.  读命令（如果没有命令发送过来则会阻塞）
10. 检查GTM Standby是否更新。 如果stanby信息更新，全局所有子线程结构信息中标志standby准备完成，该线程未连接standby，且不为GTM\_THREAD\_BACKUP状态，则该线程需要连接到该standby，如果该standby也连接不上，需要将全局所有子线程结构信息中标志standby标志为false。如果全局所有子线程结构信息中标志standby标志为false且该线程能连接到standby，该线程需要从该standby断开连接。
11. 处理命令消息，5种情况：当命令类型为C时，调用ProcessCommand函数；当命令类型为X时，移除所有事务信息；当读到EOF时，是连接断开请求；当命令类型为F时，将所有数据进行刷新；默认移除客户端打开的所有事务。

void *GTM_ThreadMain(void *argp) { 
    	GTM_ThreadInfo		*thrinfo = (GTM_ThreadInfo *)argp;
    	int 				qtype;
    	StringInfoData 		input_message;
    	sigjmp_buf  		local_sigjmp_buf;
    	char				startup_type;
    	GTM_StartupPacket	sp;
    	StringInfoData		inBuf;
    	StringInfoData		buf;
    	elog(DEBUG3, "Starting the connection helper thread");
    	MessageContext = AllocSetContextCreate(TopMemoryContext,"MessageContext",ALLOCSET_DEFAULT_MINSIZE,ALLOCSET_DEFAULT_INITSIZE,ALLOCSET_DEFAULT_MAXSIZE,false);
    	GTM_RWLockAcquire(&thrinfo->thr_lock, GTM_LOCKMODE_WRITE);
    	startup_type = pq_getbyte(thrinfo->thr_conn->con_port);
    	if (startup_type != 'A'){ 
    		ereport(ERROR,(EPROTO,errmsg("Expecting a startup message, but received %c",startup_type)));
    	}
    	initStringInfo(&inBuf);
    	/* All frontend messages have a length word next after the type code; we can read the message contents independently of the type. */
    	if (pq_getmessage(thrinfo->thr_conn->con_port, &inBuf, 0))
    		ereport(ERROR,(EPROTO,errmsg("Expecting coordinator ID, but received EOF")));
    	memcpy(&sp,pq_getmsgbytes(&inBuf, sizeof (GTM_StartupPacket)),sizeof (GTM_StartupPacket));
    	pq_getmsgend(&inBuf);
    	GTM_RegisterPGXCNode(thrinfo->thr_conn->con_port, sp.sp_node_name);
    	thrinfo->thr_conn->con_port->remote_type = sp.sp_remotetype;
    	thrinfo->thr_conn->con_port->is_postmaster = sp.sp_ispostmaster;
    	if ((sp.sp_client_id != 0) && (sp.sp_client_id <= GTMThreads->gt_starting_client_id)){ 
    		thrinfo->thr_client_id = sp.sp_client_id;
    	}
    	pq_beginmessage(&buf, 'R');
    	pq_sendint(&buf, thrinfo->thr_client_id, 4);
    	pq_endmessage(thrinfo->thr_conn->con_port, &buf);
    	pq_flush(thrinfo->thr_conn->con_port);
    	elog(DEBUG3, "Sent connection authentication message to the client");
    	/* Get the input_message in the TopMemoryContext so that we don't need to free/palloc it for every incoming message. Unlike Postgres, we don't expect the incoming messages to be of arbitrary sizes */
    	initStringInfo(&input_message);
    	
    	if (sigsetjmp(local_sigjmp_buf, 1) != 0){ 
    		/* Release all mutex and rwlocks */
    		GTM_MutexLockReleaseAll();
    		GTM_RWLockReleaseAll();
    		/* We're expected to hold this lock */
    		GTM_RWLockAcquire(&thrinfo->thr_lock, GTM_LOCKMODE_WRITE);
    		/* Report the error to the client and/or server log */
    		if (thrinfo->thr_conn)
    			EmitErrorReport(thrinfo->thr_conn->con_port);
    		else
    			EmitErrorReport(NULL);
    		/* Now return to normal top-level context and clear ErrorContext for next time. */
    		MemoryContextSwitchTo(TopMemoryContext);
    		FlushErrorState();
    	}
    	/* We can now handle ereport(ERROR) */
    	PG_exception_stack = &local_sigjmp_buf;
    
    
    	for (;;) { 
    		MemoryContextSwitchTo(MessageContext);
    		MemoryContextResetAndDeleteChildren(MessageContext);
    		resetStringInfo(&input_message);
    		GTM_RWLockRelease(&thrinfo->thr_lock);
    		qtype = ReadCommand(thrinfo->thr_conn->con_port, &input_message);
    		GTM_RWLockAcquire(&thrinfo->thr_lock, GTM_LOCKMODE_WRITE);
    		if (GTMThreads->gt_standby_ready && thrinfo->thr_conn->standby == NULL && thrinfo->thr_status != GTM_THREAD_BACKUP){ 
    			/* Connect to GTM-Standby */
    			thrinfo->thr_conn->standby = gtm_standby_connect_to_standby();
    			if (thrinfo->thr_conn->standby == NULL)
    				GTMThreads->gt_standby_ready = false;	/* This will make other threads to disconnect from the standby, if needed.*/
    		}else if (GTMThreads->gt_standby_ready == false && thrinfo->thr_conn->standby){ 
    			/* Disconnect from GTM-Standby */
    			gtm_standby_disconnect_from_standby(thrinfo->thr_conn->standby);
    			thrinfo->thr_conn->standby = NULL;
    		}
    
    		switch(qtype){ 
    			case 'C':
    				ProcessCommand(thrinfo->thr_conn->con_port, &input_message);
    				break;
    			case 'X':
    				elog(DEBUG1, "Removing all transaction infos - qtype:X");
    			case EOF:
    				/* Connection termination request * Remove all transactions opened within the thread. Note that * we don't remove transaction infos if we are a standby and * the transaction infos actually correspond to in-progress * transactions on the master */
    				elog(DEBUG1, "Removing all transaction infos - qtype:EOF");
    				if (!Recovery_IsStandby())
    					GTM_RemoveAllTransInfos(thrinfo->thr_client_id, -1);
    				/* Disconnect node if necessary */
    				Recovery_PGXCNodeDisconnect(thrinfo->thr_conn->con_port);
    				GTM_RWLockRelease(&thrinfo->thr_lock);
    				pthread_exit(thrinfo);
    				break;
    			case 'F':
    				/* Flush all the outgoing data on the wire. Consume the message * type field for sanity */
    				/* Sync with standby first */
    				if (thrinfo->thr_conn->standby){ 
    					if (Backup_synchronously)
    						gtm_sync_standby(thrinfo->thr_conn->standby);
    					else
    						gtmpqFlush(thrinfo->thr_conn->standby);
    				}
    				pq_getmsgint(&input_message, sizeof (GTM_MessageType));
    				pq_getmsgend(&input_message);
    				pq_flush(thrinfo->thr_conn->con_port);
    				break;
    			default:
    				/* Remove all transactions opened by the client */
    				GTM_RemoveAllTransInfos(thrinfo->thr_client_id, -1);
    				/* Disconnect node if necessary */
    				Recovery_PGXCNodeDisconnect(thrinfo->thr_conn->con_port);
    				ereport(FATAL,(EPROTO,errmsg("invalid frontend message type %d",qtype)));
    				break;
    		}
    	}
    	/* can't get here because the above loop never exits */
    	Assert(false);
    	return thrinfo;
    }

如下为客户端连接GTM Master的整个操作流程：  
客户端 – connect --> GTM Master  
GTM Master 主进程 – do – GTMAddConnection – fork – GTM Master 子线程  
客户端 --> GTM\_StartupPacket --> GTM Master 子线程  
GTM Master 子线程 – do – GTM\_RegisterPGXCNode  
GTM Master 子线程 – dummy认证请求信息’R’ --> 客户端  
GTM Master 子线程 <-- ReadCommand – 客户端

GTM\_ThreadMain()调用ProcessCommand()来处理来自GTM客户端的“command”消息。ProcessCommand()调度消息类型，如下表所示。如果需要，它还会在处理消息后保存事务和序列信息。

处理功能在处理消息的同时发送命令消息，**向GTM slave发送备份消息**，并向分配的GTM客户端发送响应消息。如果客户端是后端，则会立即发送响应消息。但如果是GTM proxy，则该函数只将响应消息放入libpq buffer，并在接收到类型为“F”（flush）的消息时刷新缓冲区。这个基础对于所有的处理函数来说都是通用的，所以除非这些函数有什么重要的意义，否则不会对它们进行解释。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][2021060620415760.png]

## GTM Standby ##

在main（main.c文件）中，GTM Standby按以下顺序初始化。从GTM active恢复信息非常复杂，但请注意，基本上这些序列与active相似。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 3]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 4]

ServerLoop函数涉及GTM Master新建客户端连接并连接GTM Standby的代码部分，GTM Master会先连接GTM Standby（gtm\_standby\_connect\_to\_standby）， GTM Master断开到刚建立的到GTM Standby的连接。

gtm\_standby\_connect\_to\_standby函数用于在GTM Master获取到客户端连接时连接到GTM Standby。gtm\_standby\_connect\_to\_standby\_int函数如果在GTM Standby上运行时将会返回NULL，本身就是stanby，就不用尝试连接standby。find\_standby\_node\_info函数查找一个GTM Standby的信息，如果成功返回指向GTM\_PGXCNodeInfo结构体的指针，如果失败则返回NULL。

GTM_Conn *gtm_standby_connect_to_standby(void) { 
    	GTM_Conn *conn;
    	int report;
    	conn = gtm_standby_connect_to_standby_int(&report);
    	return conn;
    }
    
    static GTM_Conn *gtm_standby_connect_to_standby_int(int *report_needed){ 
    	GTM_Conn *standby = NULL;
    	GTM_PGXCNodeInfo *n;
    	char conn_string[1024];
    	*report_needed = 0;
    	if (Recovery_IsStandby())
    		return NULL;
    	n = find_standby_node_info();
    	if (!n){ 
    		elog(DEBUG1, "Any GTM standby node not found in registered node(s).");
    		return NULL;
    	}
    	elog(DEBUG1, "GTM standby is active. Going to connect.");
    	*report_needed = 1;
    	snprintf(conn_string, sizeof(conn_string),"host=%s port=%d node_name=%s remote_type=4",n->ipaddress, n->port, NodeName);
    	standby = PQconnectGTM(conn_string);
    	if ( !standby ){ 
    	 	elog(DEBUG1, "Failed to establish a connection with GTM standby. - %p", n);
    		return NULL;
    	}
    	elog(DEBUG1, "Connection established with GTM standby. - %p", n);
    	return standby;
    }
    
    GTM_PGXCNodeInfo *find_standby_node_info(void){ 
    	GTM_PGXCNodeInfo *node[1024];
    	size_t n;
    	int i;
    	n = pgxcnode_find_by_type(GTM_NODE_GTM, node, 1024);
    	for (i = 0 ; i < n ; i++){ 
    		elog(DEBUG1, "pgxcnode_find_by_type: nodename=%s, type=%d, ipaddress=%s, port=%d, status=%d",node[i]->nodename,node[i]->type,node[i]->ipaddress,node[i]->port,node[i]->status);
    		/* Must not try and connect to ourself. That will lead to a deadlock * !!TODO Ideally we should not be registered on the GTM, but when a * failover happens, the standby may carry forward the node * registration information previously sent by the original master as a * backup. This needs to be studied further */
    		if (strcmp(node[i]->nodename, NodeName) &&
    			node[i]->status == NODE_CONNECTED)
    			return node[i];
    	}
    	return NULL;
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70]: /images/20221005/8a94139d09ed49f782813dc18dc07fe8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/83aeb79cbbef4f9fa4aecbd8c1694083.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 1]: /images/20221005/11cac21553314e06a5ca38b70d3b5303.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 2]: /images/20221005/122adf4a29ff4fb8804d1225ee99fec4.png
[2021060620415760.png]: /images/20221005/bd6bcfa9bfcf4fc0ae9bc31542d4c301.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 3]: /images/20221005/2d8020c918e5449ba83c5200a6888e03.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70 4]: /images/20221005/d07a1149b541417baecfc5f5c71a612d.png