GUC: DMQ Heartbeat timeout

danolivo · danolivo · commit 429d2256ab33 · 2019-05-22T06:50:03.000+05:00
diff --git a/contrib/pg_exchange/dmq.c b/contrib/pg_exchange/dmq.c
@@ -129,6 +129,7 @@ static shmem_startup_hook_type PreviousShmemStartupHook;
 dmq_receiver_hook_type dmq_receiver_start_hook;
 dmq_receiver_hook_type dmq_receiver_stop_hook;
 
+int dmq_heartbeat_timeout;
 void dmq_sender_main(Datum main_arg);
 
 PG_FUNCTION_INFO_V1(dmq_receiver_loop);
@@ -1138,9 +1139,9 @@ dmq_receiver_loop(PG_FUNCTION_ARGS)
 		// XXX: is it enough?
 		CHECK_FOR_INTERRUPTS();
 
-		if (dmq_now() - last_message_at > 2000)
+		if (dmq_now() - last_message_at > dmq_heartbeat_timeout)
 		{
-			mtm_log(ERROR, "[DMQ] exit receiver due to heatbeat timeout");
+			mtm_log(ERROR, "[DMQ] exit receiver due to heartbeat timeout");
 		}
 
 	}
diff --git a/contrib/pg_exchange/dmq.h b/contrib/pg_exchange/dmq.h
@@ -19,6 +19,8 @@ typedef int8 DmqSenderId;
 #define DMQ_MAX_DESTINATIONS 127
 #define DMQ_MAX_RECEIVERS 100
 
+extern int dmq_heartbeat_timeout;
+
 extern void dmq_init(const char *library_name);
 
 extern DmqDestinationId dmq_destination_add(char *connstr,
diff --git a/contrib/pg_exchange/pg_exchange.c b/contrib/pg_exchange/pg_exchange.c
@@ -80,6 +80,18 @@ _PG_init(void)
 							 NULL,
 							 NULL);
 
+	DefineCustomIntVariable("dmq_heartbeat_timeout",
+							"Max timeout between heartbeat messages",
+							NULL,
+							&dmq_heartbeat_timeout,
+							20000,
+							1, INT_MAX,
+							PGC_USERSET,
+							GUC_UNIT_MS,
+							NULL,
+							NULL,
+							NULL);
+
 	EXCHANGE_Init_methods();
 	DUMMYSCAN_Init_methods();
 	EXEC_Hooks_init();

Original file line number	Diff line number	Diff line change
`@@ -129,6 +129,7 @@ static shmem_startup_hook_type PreviousShmemStartupHook;`
`129`	`129`	`dmq_receiver_hook_type dmq_receiver_start_hook;`
`130`	`130`	`dmq_receiver_hook_type dmq_receiver_stop_hook;`
`131`	`131`
	`132`	`+int dmq_heartbeat_timeout;`
`132`	`133`	`void dmq_sender_main(Datum main_arg);`
`133`	`134`
`134`	`135`	`PG_FUNCTION_INFO_V1(dmq_receiver_loop);`
`@@ -1138,9 +1139,9 @@ dmq_receiver_loop(PG_FUNCTION_ARGS)`
`1138`	`1139`	`// XXX: is it enough?`
`1139`	`1140`	`CHECK_FOR_INTERRUPTS();`
`1140`	`1141`
`1141`		`- if (dmq_now() - last_message_at > 2000)`
	`1142`	`+ if (dmq_now() - last_message_at > dmq_heartbeat_timeout)`
`1142`	`1143`	`{`
`1143`		`- mtm_log(ERROR, "[DMQ] exit receiver due to heatbeat timeout");`
	`1144`	`+ mtm_log(ERROR, "[DMQ] exit receiver due to heartbeat timeout");`
`1144`	`1145`	`}`
`1145`	`1146`
`1146`	`1147`	`}`