PGPRO-4197: code cleanup + send stuff by 1024 bytes now

Author: anna-akenteva

pg_query_state.c

@@ -60,6 +60,9 @@ static void qs_ExecutorRun(QueryDesc *queryDesc, ScanDirection direction,
 #endif
 static void qs_ExecutorFinish(QueryDesc *queryDesc);
 
+static shm_mq_result receive_msg_by_parts(shm_mq_handle *mqh, Size *total,
+											 void **datap, bool nowait);
+
 /* Global variables */
 List 					*QueryDescStack = NIL;
 static ProcSignalReason UserIdPollReason = INVALID_PROCSIGNAL;
@@ -777,7 +780,7 @@ shm_mq_receive_with_timeout(shm_mq_handle *mqh,
 	{
 		shm_mq_result mq_receive_result;
 
-		mq_receive_result = shm_mq_receive(mqh, nbytesp, datap, true);
+		mq_receive_result = receive_msg_by_parts(mqh, nbytesp, datap, true);
 		if (mq_receive_result != SHM_MQ_WOULD_BLOCK)
 			return mq_receive_result;
 		if (rc & WL_TIMEOUT || delay <= 0)
@@ -960,51 +963,36 @@ copy_msg(shm_mq_msg *msg)
 	return result;
 }
 
-// ----------------- DEBUG -----------------
-static void
-print_recv_bytes(int num, char *src, int offset)
-{
-	elog(INFO, "======= RECV MSG SEGMENT START (%d bytes) =======", num);
-	for (int i = offset; i < offset + num; i++)
-		elog(INFO, "RECV byte #%d = %02x", i, (unsigned char) *(src + i));
-}
-// ----------------- DEBUG -----------------
-
 static shm_mq_result
-shm_mq_receive_by_bytes(shm_mq_handle *mqh, Size *total, void **datap)
+receive_msg_by_parts(shm_mq_handle *mqh, Size *total, void **datap,
+						bool nowait)
 {
 	shm_mq_result mq_receive_result;
 	shm_mq_msg	*buff;
-	int			ii;
 	int			offset;
 	int		   *expected;
 	Size		len;
 
 	/* Get the expected number of bytes in message */
-	mq_receive_result = shm_mq_receive(mqh, &len, (void **) &expected, false);
+	mq_receive_result = shm_mq_receive(mqh, &len, (void **) &expected, nowait);
 	if (mq_receive_result != SHM_MQ_SUCCESS)
 		return mq_receive_result;
 	Assert(len == sizeof(int));
-//	elog(INFO, "======= RECV MSG (expecting %d bytes) =======", *expected);
 
 	*datap = palloc0(*expected);
 
 	/* Get the message itself */
-	for (offset = 0, ii = 0; offset < *expected; ii++)
+	for (offset = 0; offset < *expected; )
 	{
-		// Keep receiving new messages until we assemble the full message
-		mq_receive_result = shm_mq_receive(mqh, &len, ((void **) &buff), false);
+		/* Keep receiving new messages until we assemble the full message */
+		mq_receive_result = shm_mq_receive(mqh, &len, ((void **) &buff), nowait);
 		memcpy((char *) *datap + offset, buff, len);
-//		print_recv_bytes(len, (char *) *datap, offset);
 		offset += len;
 		if (mq_receive_result != SHM_MQ_SUCCESS)
 			return mq_receive_result;
 	}
 
-//	elog(INFO, "RECV: END cycle - %d", ii);
 	*total = offset;
-//	mq_receive_result = shm_mq_receive(mqh, &len, (void **) &msg, false);
-//	*datap = buff;
 
 	return mq_receive_result;
 }
@@ -1081,7 +1069,8 @@ 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_by_bytes(mqh, &len, ((void **) &msg));
+	mq_receive_result = receive_msg_by_parts(mqh, &len, (void **) &msg,
+												false);
 	if (mq_receive_result != SHM_MQ_SUCCESS)
 		goto mq_error;
 	if (msg->reqid != reqid)

pg_query_state.h

@@ -18,7 +18,7 @@
 #include "storage/shm_mq.h"
 
 #define	QUEUE_SIZE			(16 * 1024)
-#define BUF_SIZE			7
+#define MSG_MAX_SIZE		1024
 
 #define TIMINIG_OFF_WARNING 1
 #define BUFFERS_OFF_WARNING 2

signal_handler.c

@@ -27,7 +27,7 @@ typedef struct
 	char		*plan;
 } stack_frame;
 
-static void send_msg_by_bits(shm_mq_handle *mqh, Size nbytes, const void *data);
+static void send_msg_by_parts(shm_mq_handle *mqh, Size nbytes, const void *data);
 
 /*
  *	Get List of stack_frames as a stack of function calls starting from outermost call.
@@ -150,33 +150,21 @@ serialize_stack(char *dest, List *qs_stack)
 		serialize_stack_frame(&dest, qs_frame);
 	}
 }
-// ----------------- DEBUG -----------------
-static void
-print_sent_bytes(int num, char *src, int offset)
-{
-	elog(INFO, "======= SEND MSG SEGMENT START (%d bytes) =======", num);
-	for (int i = offset; i < offset + num; i++)
-		elog(INFO, "SENT byte #%d = %02x", i, (unsigned char) *(src + i));
-}
-// ----------------- DEBUG -----------------
 
 static void
-send_msg_by_bits(shm_mq_handle *mqh, Size nbytes, const void *data)
+send_msg_by_parts(shm_mq_handle *mqh, Size nbytes, const void *data)
 {
 	int bytes_left;
 	int bytes_send;
 
 	/* Send the expected message length */
 	shm_mq_send(mqh, sizeof(int), &nbytes, false);
 
-//	elog(INFO, "======= SEND MSG (%lu bytes) =======", nbytes);
 	for (int offset = 0; offset < nbytes; offset += bytes_send)
 	{
 		bytes_left = nbytes - offset;
-		bytes_send = (bytes_left < BUF_SIZE) ? bytes_left : BUF_SIZE;
+		bytes_send = (bytes_left < MSG_MAX_SIZE) ? bytes_left : MSG_MAX_SIZE;
 		shm_mq_send(mqh, bytes_send, &(((unsigned char*)data)[offset]), false);
-		// DEBUG: print message that we just sent
-//		print_sent_bytes(bytes_send, (char *) data, offset);
 	}
 }
 
@@ -238,15 +226,15 @@ SendQueryState(void)
 	{
 		shm_mq_msg msg = { reqid, BASE_SIZEOF_SHM_MQ_MSG, MyProc, STAT_DISABLED };
 
-		send_msg_by_bits(mqh, msg.length, &msg);
+		send_msg_by_parts(mqh, msg.length, &msg);
 	}
 
 	/* check if backend doesn't execute any query */
 	else if (list_length(QueryDescStack) == 0)
 	{
 		shm_mq_msg msg = { reqid, BASE_SIZEOF_SHM_MQ_MSG, MyProc, QUERY_NOT_RUNNING };
 
-		send_msg_by_bits(mqh, msg.length, &msg);
+		send_msg_by_parts(mqh, msg.length, &msg);
 	}
 
 	/* happy path */
@@ -269,7 +257,7 @@ SendQueryState(void)
 
 		msg->stack_depth = list_length(qs_stack);
 		serialize_stack(msg->stack, qs_stack);
-		send_msg_by_bits(mqh, msglen, msg);
+		send_msg_by_parts(mqh, msglen, msg);
 	}
 	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();