Merge commit '3551603141e8d285c6bfc302831bd095c05ce872' into PGPROEE9_6_wait_sampling

Author: vbwagner

collector.c

@@ -356,15 +356,19 @@ collector_main(Datum main_arg)
 					profile_period - (int)profile_diff));
 
 		if (rc & WL_POSTMASTER_DEATH)
-			exit(1);
+			proc_exit(1);
 
 		ResetLatch(&MyProc->procLatch);
 
 		/* Handle request if any */
 		if (collector_hdr->request != NO_REQUEST)
 		{
-			SHMRequest request = collector_hdr->request;
+			LOCKTAG		tag;
+			SHMRequest	request = collector_hdr->request;
 
+			init_lock_tag(&tag, PGWS_COLLECTOR_LOCK);
+
+			LockAcquire(&tag, ExclusiveLock, false, false);
 			collector_hdr->request = NO_REQUEST;
 
 			if (request == HISTORY_REQUEST || request == PROFILE_REQUEST)
@@ -392,6 +396,7 @@ collector_main(Datum main_arg)
 				hash_destroy(profile_hash);
 				profile_hash = make_profile_hash();
 			}
+			LockRelease(&tag, ExclusiveLock, false);
 		}
 	}
 

pg_wait_sampling.c

@@ -414,11 +414,11 @@ typedef struct
 	ProfileItem	   *items;
 } Profile;
 
-static void
-init_lock_tag(LOCKTAG *tag)
+void
+init_lock_tag(LOCKTAG *tag, uint32 lock)
 {
 	tag->locktag_field1 = PG_WAIT_SAMPLING_MAGIC;
-	tag->locktag_field2 = 0;
+	tag->locktag_field2 = lock;
 	tag->locktag_field3 = 0;
 	tag->locktag_field4 = 0;
 	tag->locktag_type = LOCKTAG_USERLOCK;
@@ -428,7 +428,8 @@ init_lock_tag(LOCKTAG *tag)
 static void *
 receive_array(SHMRequest request, Size item_size, Size *count)
 {
-	LOCKTAG			tag;
+	LOCKTAG			queueTag;
+	LOCKTAG			collectorTag;
 	shm_mq		   *mq;
 	shm_mq_handle  *mqh;
 	shm_mq_result	res;
@@ -438,8 +439,14 @@ receive_array(SHMRequest request, Size item_size, Size *count)
 	Pointer			result,
 					ptr;
 
-	init_lock_tag(&tag);
-	LockAcquire(&tag, ExclusiveLock, false, false);
+	/* Ensure nobody else trying to send request to queue */
+	init_lock_tag(&queueTag, PGWS_QUEUE_LOCK);
+	LockAcquire(&queueTag, ExclusiveLock, false, false);
+
+	/* Ensure collector has processed previous request */
+	init_lock_tag(&collectorTag, PGWS_COLLECTOR_LOCK);
+	LockAcquire(&collectorTag, ExclusiveLock, false, false);
+	LockRelease(&collectorTag, ExclusiveLock, false);
 
 	mq = shm_mq_create(collector_mq, COLLECTOR_QUEUE_SIZE);
 	collector_hdr->request = request;
@@ -472,7 +479,7 @@ receive_array(SHMRequest request, Size item_size, Size *count)
 
 	shm_mq_detach(mq);
 
-	LockRelease(&tag, ExclusiveLock, false);
+	LockRelease(&queueTag, ExclusiveLock, false);
 
 	return result;
 }
@@ -568,13 +575,18 @@ Datum
 pg_wait_sampling_reset_profile(PG_FUNCTION_ARGS)
 {
 	LOCKTAG		tag;
+	LOCKTAG		tagCollector;
 
 	check_shmem();
 
-	init_lock_tag(&tag);
+	init_lock_tag(&tag, false);
 
 	LockAcquire(&tag, ExclusiveLock, false, false);
 
+	init_lock_tag(&tagCollector, true);
+	LockAcquire(&tagCollector, ExclusiveLock, false, false);
+	LockRelease(&tagCollector, ExclusiveLock, false);
+
 	collector_hdr->request = PROFILE_RESET;
 	SetLatch(collector_hdr->latch);
 

pg_wait_sampling.h

@@ -24,6 +24,8 @@
 #define	PG_WAIT_SAMPLING_MAGIC		0xCA94B107
 #define COLLECTOR_QUEUE_SIZE		(16 * 1024)
 #define HISTORY_TIME_MULTIPLIER		10
+#define PGWS_QUEUE_LOCK				0
+#define PGWS_COLLECTOR_LOCK			1
 
 typedef struct
 {
@@ -70,6 +72,7 @@ extern void check_shmem(void);
 extern CollectorShmqHeader *collector_hdr;
 extern shm_mq			   *collector_mq;
 extern void read_current_wait(PGPROC *proc, HistoryItem *item);
+extern void init_lock_tag(LOCKTAG *tag, uint32 lock);
 
 /* collector.c */
 extern void register_wait_collector(void);