Fix the issue about renumbering a connection id in GTM proxy.

When a backend disconnect from a GTM proxy, the proxy renumbers a connection id of the connection located in last slot of a connection array. GTM proxy doesn't notify
that action to a GTM. Then a GTM could release unrelated transaction ids.

Author: SAITO Masataka

src/gtm/proxy/proxy_main.c

@@ -136,6 +136,9 @@ static int ServerLoop(void);
 static int initMasks(fd_set *rmask);
 void *GTMProxy_ThreadMain(void *argp);
 static int GTMProxyAddConnection(Port *port);
+static GTMProxy_ConnectionInfo *GTMProxy_GetConnInfo(GTMProxy_ThreadInfo *thrinfo,
+													 GTMProxy_ConnID con_id);
+static int GTMProxy_GetConnInfoIndex(GTMProxy_ThreadInfo *thrinfo, GTMProxy_ConnID con_id);
 static int ReadCommand(GTMProxy_ConnectionInfo *conninfo, StringInfo inBuf);
 static void GTMProxy_HandshakeConnection(GTMProxy_ConnectionInfo *conninfo);
 static void GTMProxy_HandleDisconnect(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn);
@@ -1554,6 +1557,27 @@ GTMProxyAddConnection(Port *port)
 	return STATUS_OK;
 }
 
+/* Convert a connection id to a index in GTMProxy_ThreadInfo::thr_all_conns */
+static int
+GTMProxy_GetConnInfoIndex(GTMProxy_ThreadInfo *thrinfo, GTMProxy_ConnID con_id)
+{
+	if (con_id == InvalidGTMProxyConnID)
+		return -1;
+	return thrinfo->thr_conid2idx[con_id];
+}
+
+static GTMProxy_ConnectionInfo *
+GTMProxy_GetConnInfo(GTMProxy_ThreadInfo *thrinfo, GTMProxy_ConnID con_id)
+{
+	int con_idx;
+
+	con_idx = GTMProxy_GetConnInfoIndex(thrinfo, con_id);
+	if (con_idx < 0)
+		return NULL;
+
+	return thrinfo->thr_all_conns[con_idx];
+}
+
 void
 ProcessCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn,
 		StringInfo input_message)
@@ -1684,7 +1708,9 @@ HandleGTMError(GTM_Conn *gtm_conn)
 static GTM_Conn *
 HandlePostCommand(GTMProxy_ConnectionInfo *conninfo, GTM_Conn *gtm_conn)
 {
-	int		connIdx = conninfo->con_id;
+	int connIdx;
+
+	connIdx = GTMProxy_GetConnInfoIndex(GetMyThreadInfo, conninfo->con_id);
 
 	Assert(conninfo && gtm_conn);
 	/*
@@ -1896,12 +1922,15 @@ ProcessResponse(GTMProxy_ThreadInfo *thrinfo, GTMProxy_CommandInfo *cmdinfo,
 		case MSG_SEQUENCE_CLOSE:
 		case MSG_SEQUENCE_RENAME:
 		case MSG_SEQUENCE_ALTER:
-			if ((res->gr_proxyhdr.ph_conid == InvalidGTMProxyConnID) ||
-				(res->gr_proxyhdr.ph_conid >= GTM_PROXY_MAX_CONNECTIONS) ||
-				(thrinfo->thr_all_conns[res->gr_proxyhdr.ph_conid] != cmdinfo->ci_conn))
 			{
-				ReleaseCmdBackup(cmdinfo);
-				elog(PANIC, "Invalid response or synchronization loss");
+				GTMProxy_ConnID con_id = res->gr_proxyhdr.ph_conid;
+				if ((con_id == InvalidGTMProxyConnID) ||
+					(con_id >= GTM_PROXY_MAX_CONNECTIONS) ||
+					(GTMProxy_GetConnInfo(thrinfo, con_id) != cmdinfo->ci_conn))
+				{
+					ReleaseCmdBackup(cmdinfo);
+					elog(PANIC, "Invalid response or synchronization loss");
+				}
 			}
 
 			/*
@@ -1975,9 +2004,10 @@ static int
 ReadCommand(GTMProxy_ConnectionInfo *conninfo, StringInfo inBuf)
 {
 	int 			qtype;
-	int				connIdx = conninfo->con_id;
+	int				connIdx;
 	int				anyBackup;
 
+	connIdx = GTMProxy_GetConnInfoIndex(GetMyThreadInfo, conninfo->con_id);
 	anyBackup = (GetMyThreadInfo->thr_any_backup[connIdx] ? TRUE : FALSE);
 
 	/*
@@ -3327,7 +3357,9 @@ ConnectGTM(void)
  */
 static void ReleaseCmdBackup(GTMProxy_CommandInfo *cmdinfo)
 {
-	GTMProxy_ConnID connIdx = cmdinfo->ci_conn->con_id;
+	GTMProxy_ConnID connIdx;
+
+	connIdx = GTMProxy_GetConnInfoIndex(GetMyThreadInfo, cmdinfo->ci_conn->con_id);
 
 	GetMyThreadInfo->thr_any_backup[connIdx] = FALSE;
 	GetMyThreadInfo->thr_qtype[connIdx] = 0;

src/gtm/proxy/proxy_thread.c

@@ -144,6 +144,9 @@ GTMProxy_ThreadCreate(void *(* startroutine)(void *), int idx)
 	GTM_MutexLockInit(&thrinfo->thr_lock);
 	GTM_CVInit(&thrinfo->thr_cv);
 
+	/* Initialize mapping to be unassigned. */
+	memset(thrinfo->thr_conid2idx, 0xff, sizeof(thrinfo->thr_conid2idx));
+
 	/*
 	 * Initialize communication area with SIGUSR2 signal handler (reconnect)
 	 */
@@ -331,6 +334,7 @@ GTMProxy_ThreadMainWrapper(void *argp)
 GTMProxy_ThreadInfo *
 GTMProxy_ThreadAddConnection(GTMProxy_ConnectionInfo *conninfo)
 {
+	int con_id = -1, con_idx = 0;
 	GTMProxy_ThreadInfo *thrinfo = NULL;
 
 	/*
@@ -368,15 +372,27 @@ GTMProxy_ThreadAddConnection(GTMProxy_ConnectionInfo *conninfo)
 		elog(ERROR, "Too many connections");
 	}
 
+	/* Find unassigned connection id in this worker thread. */
+	for (con_id = 0; con_id < GTM_PROXY_MAX_CONNECTIONS; con_id++)
+		if (thrinfo->thr_conid2idx[con_id] == -1)
+			break;
+
+	if (con_id >= GTM_PROXY_MAX_CONNECTIONS) {
+		GTM_MutexLockRelease(&thrinfo->thr_lock);
+		elog(ERROR, "Unassigned connection id not found.");
+	}
+
 	/*
 	 * Save the array slotid in the conninfo structure. We send this to the GTM
 	 * server as an identifier which the GTM server sends us back in the
 	 * response. We use that information to route the response back to the
 	 * approrpiate connection
 	 */
-	conninfo->con_id = thrinfo->thr_conn_count;
-	thrinfo->thr_all_conns[thrinfo->thr_conn_count] = conninfo;
-	thrinfo->thr_conn_count++;
+	con_idx = thrinfo->thr_conn_count++;
+	conninfo->con_id = con_id;
+	thrinfo->thr_conid2idx[con_id] = con_idx;
+	thrinfo->thr_all_conns[con_idx] = conninfo;
+	elog(DEBUG5, "Assigned a connection id to new connection: id = %d, index = %d", con_id, con_idx);
 
 	/*
 	 * Now increment the seqno since a new connection is added to the array.
@@ -432,20 +448,35 @@ GTMProxy_ThreadRemoveConnection(GTMProxy_ThreadInfo *thrinfo, GTMProxy_Connectio
 	thrinfo->thr_qtype[ii] = 0;
 	resetStringInfo(&(thrinfo->thr_inBufData[ii]));
 
+	/* Release connection id */
+	if (conninfo->con_id != InvalidGTMProxyConnID)
+	{
+		thrinfo->thr_conid2idx[conninfo->con_id] = -1;
+		elog(DEBUG5, "Released connection id %d", conninfo->con_id);
+	}
+
 	/*
 	 * If this is the last entry in the array ? If not, then copy the last
 	 * entry in this slot and mark the last slot an empty
 	 */
 	if ((ii + 1) < thrinfo->thr_conn_count)
 	{
+		GTMProxy_ConnectionInfo *ci_moved;
+		int last_idx;
+
+		/* Pick up last slot */
+		last_idx = thrinfo->thr_conn_count - 1;
+		ci_moved = thrinfo->thr_all_conns[last_idx];
+		
 		/* Copy the last entry in this slot */
-		thrinfo->thr_all_conns[ii] = thrinfo->thr_all_conns[thrinfo->thr_conn_count - 1];
+		thrinfo->thr_all_conns[ii] = ci_moved;
 
 		/* Mark the last slot free */
-		thrinfo->thr_all_conns[thrinfo->thr_conn_count - 1] = NULL;
+		thrinfo->thr_all_conns[last_idx] = NULL;
 
-		/* Adjust the con_id to reflect the current slot in the array */
-		thrinfo->thr_all_conns[ii]->con_id = ii;
+		/* Adjust the mapping to reflect the current slot in the array */
+		if (ci_moved->con_id != InvalidGTMProxyConnID)
+			thrinfo->thr_conid2idx[ci_moved->con_id] = ii;
 	}
 	else
 	{

src/include/gtm/gtm_proxy.h

@@ -111,6 +111,9 @@ typedef struct GTMProxy_ThreadInfo
 	GTMProxy_ConnectionInfo	*thr_all_conns[GTM_PROXY_MAX_CONNECTIONS];
 	struct pollfd			thr_poll_fds[GTM_PROXY_MAX_CONNECTIONS];
 
+	/* map info from ConnectionInfo->con_id to array index of thr_all_conns */
+	int16					thr_conid2idx[GTM_PROXY_MAX_CONNECTIONS];
+
 	/* Command backup */
 	short					thr_any_backup[GTM_PROXY_MAX_CONNECTIONS];
 	int						thr_qtype[GTM_PROXY_MAX_CONNECTIONS];