Postgres-xl GTM（全局事务管理器 Globale Transaction Manager） Node 管理器

刺骨的言语ヽ痛彻心扉 2022-10-05 05:51 162阅读 0赞

# Node Manager #

节点数据结构主要成员：节点类型（Proxy、Coordinator、Datanode），节点序号、Proxy序号（如果节点不经过proxy或节点类型为proxy，该成员设置为0），Postgresql端口、GTM可见的UP，节点的数据目录。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center]

void GTM_InitNodeManager(void);
    size_t pgxcnode_get_all(GTM_PGXCNodeInfo **data, size_t maxlen, bool locked);
    size_t pgxcnode_find_by_type(GTM_PGXCNodeType type, GTM_PGXCNodeInfo **data, size_t maxlen);
    
    int Recovery_PGXCNodeRegister(GTM_PGXCNodeType	type, char *nodename, GTM_PGXCNodePort port, char			*proxyname, GTM_PGXCNodeStatus	status, char *ipaddress, char *datafolder, bool	in_recovery, int			socket, bool is_session);
    int Recovery_PGXCNodeUnregister(GTM_PGXCNodeType type, char *node_name, bool in_recovery, int socket);
    void Recovery_SaveRegisterInfo(void);
    void Recovery_PGXCNodeDisconnect(Port *myport);
    void Recovery_SaveRegisterFileName(char *dir);
    int Recovery_PGXCNodeRegisterCoordProcess(char *coord_node, int coord_procid, int coord_backendid);
    
    void ProcessPGXCRegisterSession(Port *myport, StringInfo message);
    void ProcessPGXCNodeRegister(Port *myport, StringInfo message, bool is_backup);
    void ProcessPGXCNodeUnregister(Port *myport, StringInfo message, bool is_backup);
    void ProcessPGXCNodeBackendDisconnect(Port *myport, StringInfo message);
    void ProcessPGXCNodeList(Port *myport, StringInfo message);
    
    void ProcessGTMBeginBackup(Port *myport, StringInfo message);
    void ProcessGTMEndBackup(Port *myport, StringInfo message);
    
    
    GlobalTransactionId GTM_HandleGlobalXmin(GTM_PGXCNodeType type, char *node_name, GlobalTransactionId reported_xmin, int *errcode);

## 节点管理器 ##

GTM\_InitNodeManager函数初始化GTM\_PGXCNodes桶中的gtm\_list指针为null，并初始化RegisterFileLock和PGXCNodesLock锁。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70]

void GTM_InitNodeManager(void) { 
    	int ii;
    	for (ii = 0; ii < NODE_HASH_TABLE_SIZE; ii++) { 
    		GTM_PGXCNodes[ii].nhb_list = gtm_NIL;
    	}
    	GTM_RWLockInit(&RegisterFileLock);
    	GTM_RWLockInit(&PGXCNodesLock);
    }

pgxcnode\_get\_all函数将GTM\_NodeInfoHashBucket桶中gtm\_List中存储的所有GTMPGXCNodeInfo都存放到GTM\_PGXCNodeInfo \*\*data所指向的内存空间中。pgxcnode\_find\_by\_type函数查找相同GTM\_PGXCNodeType的所有节点信息，存放到GTM\_PGXCNodeInfo \*\*data所指向的内存空间中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center 1]

size_t pgxcnode_get_all(GTM_PGXCNodeInfo **data, size_t maxlen, bool locked) { 
    	GTM_PGXCNodeInfoHashBucket *bucket;
    	gtm_ListCell *elem;
    	int node = 0;
    	int i;
    	if (!locked)
    		GTM_RWLockAcquire(&PGXCNodesLock, GTM_LOCKMODE_READ);
    	for (i = 0; i < NODE_HASH_TABLE_SIZE; i++) { 
    		bucket = &GTM_PGXCNodes[i];
    		gtm_foreach(elem, bucket->nhb_list){ 
    			GTM_PGXCNodeInfo *curr_nodeinfo = NULL;
    			curr_nodeinfo = (GTM_PGXCNodeInfo *) gtm_lfirst(elem);
    			if (curr_nodeinfo != NULL) { 
    				data[node] = curr_nodeinfo;
    				node++;
    			}
    			if (node == maxlen)
    				break;
    		}
    	}	
    	if (!locked)
    		GTM_RWLockRelease(&PGXCNodesLock);
    	return node;
    }
    
    size_t pgxcnode_find_by_type(GTM_PGXCNodeType type, GTM_PGXCNodeInfo **data, size_t maxlen) { 
    	GTM_PGXCNodeInfoHashBucket *bucket;
    	gtm_ListCell *elem;
    	int node = 0;
    	int i;
    	GTM_RWLockAcquire(&PGXCNodesLock, GTM_LOCKMODE_READ);
    	for (i = 0; i < NODE_HASH_TABLE_SIZE; i++){ 
    		bucket = &GTM_PGXCNodes[i];
    		gtm_foreach(elem, bucket->nhb_list){ 
    			GTM_PGXCNodeInfo *cur = NULL;
    			cur = (GTM_PGXCNodeInfo *) gtm_lfirst(elem);
    			if (cur != NULL && cur->type == type){ 
    				data[node] = cur;
    				elog(DEBUG1, "pgxcnode_find_by_type: cur=%p, ipaddress=%s", cur, cur->ipaddress);
    				node++;
    			}
    			if (node == maxlen)
    				break;
    		}
    	}
    	GTM_RWLockRelease(&PGXCNodesLock);
    	return node;
    }

## 节点注册 ##

### 节点Recovery注册 ###

register\_common.c  
向GTM\_PGXCNODES全局哈希桶中注册GTM\_PGXCNodeInfo： int Recovery\_PGXCNodeRegister(GTM\_PGXCNodeType type, char \*nodename, GTM\_PGXCNodePort port, char \*proxyname, GTM\_PGXCNodeStatus status, char \*ipaddress, char \*datafolder, bool in\_recovery, int socket, bool is\_session);  
向GTM\_PGXCNODES全局哈希桶中取消注册GTM\_PGXCNodeInfo：int Recovery\_PGXCNodeUnregister(GTM\_PGXCNodeType type, char \*node\_name, bool in\_recovery, int socket);  
当GTM关闭时调用，向磁盘中写入注册信息（GTM\_PGXCNODES全局哈希桶中注册的节点信息）：void Recovery\_SaveRegisterInfo(void);  
关闭已经和GTM master断开连接的节点的连接：void Recovery\_PGXCNodeDisconnect(Port \*myport);  
void Recovery\_SaveRegisterFileName(char \*dir);  
注册active分布session：int Recovery\_PGXCNodeRegisterCoordProcess(char \*coord\_node, int coord\_procid, int coord\_backendid);

Recovery\_PGXCNodeRegister函数创建GTM\_PGXCNodeInfo结构体，并利用形参进行成员的填充，将该PGXC Node Info结构体加入到全局hasn table（GTM\_PGXCNODES）中。最后向磁盘上的PGXC节点文件添加注册记录。

int Recovery_PGXCNodeRegister(GTM_PGXCNodeType	type, char *nodename, GTM_PGXCNodePort port, char *proxyname, GTM_PGXCNodeStatus	status, char *ipaddress, char *datafolder, bool	in_recovery, int socket, bool is_session) { 
    	GTM_PGXCNodeInfo *nodeinfo = NULL;
    	int errcode = 0;
    	nodeinfo = (GTM_PGXCNodeInfo *) palloc0(sizeof(GTM_PGXCNodeInfo));
    	if (nodeinfo == NULL)
    		ereport(ERROR, (ENOMEM, errmsg("Out of memory")));
    	GTM_RWLockInit(&nodeinfo->node_lock);
    	/* Fill in structure */
    	nodeinfo->type = type;
    	if (nodename)
    		nodeinfo->nodename = pgxcnode_copy_char(nodename);
    	nodeinfo->port = port;
    	if (proxyname)
    		nodeinfo->proxyname = pgxcnode_copy_char(proxyname);
    	if (datafolder)
    		nodeinfo->datafolder = pgxcnode_copy_char(datafolder);
    	if (ipaddress)
    		nodeinfo->ipaddress = pgxcnode_copy_char(ipaddress);
    	nodeinfo->status = status;
    	nodeinfo->socket = socket;
    	nodeinfo->reported_xmin = InvalidGlobalTransactionId;
    	nodeinfo->reported_xmin_time = 0;
    	nodeinfo->is_session = is_session;
    	elog(DEBUG1, "Recovery_PGXCNodeRegister Request info: type=%d, nodename=%s, port=%d, datafolder=%s, ipaddress=%s, status=%d", type, nodename, port, datafolder, ipaddress, status);
    	elog(DEBUG1, "Recovery_PGXCNodeRegister Node info: type=%d, nodename=%s, port=%d, datafolder=%s, ipaddress=%s, status=%d", nodeinfo->type, nodeinfo->nodename, nodeinfo->port, nodeinfo->datafolder, nodeinfo->ipaddress, nodeinfo->status);
    	/* Add PGXC Node Info to the global hash table */
    	errcode = pgxcnode_add_info(nodeinfo);
    	/* Add a Record to file disk saying that this node with given data has been correctly registered */
    	if (!in_recovery && errcode == 0)
    		Recovery_RecordRegisterInfo(nodeinfo, true);
    	return errcode;
    }

### 节点注册处理 ###

register\_gtm.c  
处理MSG\_REGISTER\_SESSION消息：void ProcessPGXCRegisterSession(Port \*myport, StringInfo message);  
void ProcessPGXCNodeRegister(Port \*myport, StringInfo message, bool is\_backup);  
处理MSG\_NODE\_UNREGISTER/MSG\_BKUP\_NODE\_UNREGISTER消息：void ProcessPGXCNodeUnregister(Port \*myport, StringInfo message, bool is\_backup);  
处理MSG\_BACKEND\_DISCONNECT消息：void ProcessPGXCNodeBackendDisconnect(Port \*myport, StringInfo message);  
处理MSG\_NODE\_LIST消息：void ProcessPGXCNodeList(Port \*myport, StringInfo message);

void ProcessGTMBeginBackup(Port \*myport, StringInfo message);  
void ProcessGTMEndBackup(Port \*myport, StringInfo message);

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/0dd3a436bdf141ca98829ed355326233.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70]: /images/20221005/296fd194b01d433fbaf57f45d79bccf8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FzbWFydGtpbGxlcg_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221005/b536e498314141868043c63b43dbe87b.png