PGPRO-4197: attempt to send stuff byte-wise in pg_query_state

Author: anna-akenteva

pg_query_state.c

@@ -960,6 +960,55 @@ 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)
+{
+	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);
+	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++)
+	{
+		// Keep receiving new messages until we assemble the full message
+		mq_receive_result = shm_mq_receive(mqh, &len, ((void **) &buff), false);
+		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;
+}
+
 static List *
 GetRemoteBackendQueryStates(PGPROC *leader,
 							List *pworkers,
@@ -1032,7 +1081,7 @@ 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);
+	mq_receive_result = shm_mq_receive_by_bytes(mqh, &len, ((void **) &msg));
 	if (mq_receive_result != SHM_MQ_SUCCESS)
 		goto mq_error;
 	if (msg->reqid != reqid)

pg_query_state.h

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

signal_handler.c

@@ -27,6 +27,8 @@ typedef struct
 	char		*plan;
 } stack_frame;
 
+static void send_msg_by_bits(shm_mq_handle *mqh, Size nbytes, const void *data);
+
 /*
  *	Get List of stack_frames as a stack of function calls starting from outermost call.
  *		Each entry contains query text and query state in form of EXPLAIN ANALYZE output.
@@ -148,6 +150,35 @@ 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)
+{
+	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;
+		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);
+	}
+}
 
 /*
  * Send state of current query to shared queue.
@@ -207,15 +238,15 @@ SendQueryState(void)
 	{
 		shm_mq_msg msg = { reqid, BASE_SIZEOF_SHM_MQ_MSG, MyProc, STAT_DISABLED };
 
-		shm_mq_send(mqh, msg.length, &msg, false);
+		send_msg_by_bits(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 };
 
-		shm_mq_send(mqh, msg.length, &msg, false);
+		send_msg_by_bits(mqh, msg.length, &msg);
 	}
 
 	/* happy path */
@@ -238,7 +269,7 @@ SendQueryState(void)
 
 		msg->stack_depth = list_length(qs_stack);
 		serialize_stack(msg->stack, qs_stack);
-		shm_mq_send(mqh, msglen, msg, false);
+		send_msg_by_bits(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();