[refer #PGPRO-4197] Add request ID

Author: knizhnik

pg_query_state.c

@@ -952,6 +952,7 @@ GetRemoteBackendQueryStates(PGPROC *leader,
 	shm_mq_result	 mq_receive_result;
 	shm_mq_msg		*msg;
 	Size			 len;
+	static int       reqid = 0;
 
 	Assert(QueryStatePollReason != INVALID_PROCSIGNAL);
 	Assert(mq);
@@ -963,6 +964,7 @@ GetRemoteBackendQueryStates(PGPROC *leader,
 	params->buffers = buffers;
 	params->triggers = triggers;
 	params->format = format;
+	params->reqid = ++reqid;
 	pg_write_barrier();
 
 	/* initialize message queue that will transfer query states */
@@ -1002,9 +1004,13 @@ GetRemoteBackendQueryStates(PGPROC *leader,
 
 	/* extract query state from leader process */
 	mqh = shm_mq_attach(mq, NULL, NULL);
+	elog(DEBUG1, "Wait response from leader %d", leader->pid);
 	mq_receive_result = shm_mq_receive(mqh, &len, (void **) &msg, false);
 	if (mq_receive_result != SHM_MQ_SUCCESS)
 		goto mq_error;
+	if (msg->reqid != reqid)
+		goto mq_error;
+
 	Assert(len == msg->length);
 	result = lappend(result, copy_msg(msg));
 #if PG_VERSION_NUM < 100000
@@ -1021,6 +1027,7 @@ GetRemoteBackendQueryStates(PGPROC *leader,
 		PGPROC 			*proc = (PGPROC *) lfirst(iter);
 
 		/* prepare message queue to transfer data */
+		elog(DEBUG1, "Wait response from worker %d", proc->pid);
 		mq = shm_mq_create(mq, QUEUE_SIZE);
 		shm_mq_set_sender(mq, proc);
 		shm_mq_set_receiver(mq, MyProc);	/* this function notifies the
@@ -1034,9 +1041,12 @@ GetRemoteBackendQueryStates(PGPROC *leader,
 														(void **) &msg,
 														MAX_RCV_TIMEOUT);
 		if (mq_receive_result != SHM_MQ_SUCCESS)
+		{
 			/* counterpart is died, not consider it */
-			continue;
-
+			goto mq_error;
+		}
+		if (msg->reqid != reqid)
+			goto mq_error;
 		Assert(len == msg->length);
 
 		/* aggregate result data */
@@ -1054,6 +1064,11 @@ GetRemoteBackendQueryStates(PGPROC *leader,
 	ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR),
 				errmsg("invalid send signal")));
 mq_error:
+#if PG_VERSION_NUM < 100000
+	shm_mq_detach(mq);
+#else
+	shm_mq_detach(mqh);
+#endif
 	ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR),
 				errmsg("error in message queue data transmitting")));
 }

pg_query_state.h

@@ -42,6 +42,7 @@ typedef enum
  */
 typedef struct
 {
+	int     reqid;
 	int		length;							/* size of message record, for sanity check */
 	PGPROC	*proc;
 	PG_QS_RequestResult	result_code;
@@ -56,6 +57,7 @@ typedef struct
 /* pg_query_state arguments */
 typedef struct
 {
+	int     reqid;
 	bool 	verbose;
 	bool	costs;
 	bool	timing;

signal_handler.c

@@ -160,6 +160,7 @@ SendQueryState(void)
 	instr_time	start_time;
 	instr_time	cur_time;
 	int64 		delay = MAX_SND_TIMEOUT;
+	int         reqid = params->reqid;
 
 	INSTR_TIME_SET_CURRENT(start_time);
 
@@ -189,21 +190,21 @@ SendQueryState(void)
 		CHECK_FOR_INTERRUPTS();
 		ResetLatch(MyLatch);
 	}
-
+	elog(DEBUG1, "Worker %d receives pg_query_state request from %d", shm_mq_get_sender(mq)->pid, shm_mq_get_receiver(mq)->pid);
 	mqh = shm_mq_attach(mq, NULL, NULL);
 
 	/* check if module is enabled */
 	if (!pg_qs_enable)
 	{
-		shm_mq_msg msg = { BASE_SIZEOF_SHM_MQ_MSG, MyProc, STAT_DISABLED };
+		shm_mq_msg msg = { reqid, BASE_SIZEOF_SHM_MQ_MSG, MyProc, STAT_DISABLED };
 
 		shm_mq_send(mqh, msg.length, &msg, false);
 	}
 
 	/* check if backend doesn't execute any query */
 	else if (list_length(QueryDescStack) == 0)
 	{
-		shm_mq_msg msg = { BASE_SIZEOF_SHM_MQ_MSG, MyProc, QUERY_NOT_RUNNING };
+		shm_mq_msg msg = { reqid, BASE_SIZEOF_SHM_MQ_MSG, MyProc, QUERY_NOT_RUNNING };
 
 		shm_mq_send(mqh, msg.length, &msg, false);
 	}
@@ -215,6 +216,7 @@ SendQueryState(void)
 		int				msglen = sizeof(shm_mq_msg) + serialized_stack_length(qs_stack);
 		shm_mq_msg		*msg = palloc(msglen);
 
+		msg->reqid = reqid;
 		msg->length = msglen;
 		msg->proc = MyProc;
 		msg->result_code = QS_RETURNED;
@@ -229,5 +231,6 @@ SendQueryState(void)
 		serialize_stack(msg->stack, qs_stack);
 		shm_mq_send(mqh, msglen, msg, false);
 	}
+	elog(DEBUG1, "Worker %d sends response for pg_query_state to %d", shm_mq_get_sender(mq)->pid, shm_mq_get_receiver(mq)->pid);
 	DetachPeer();
 }