Start dynamic pool workers

Author: knizhnik

bgwpool.c

@@ -7,22 +7,15 @@
 #include "storage/spin.h"
 #include "storage/pg_sema.h"
 #include "storage/shmem.h"
+#include "datatype/timestamp.h"
 
 #include "bgwpool.h"
 
 bool MtmIsLogicalReceiver;
+int  MtmMaxWorkers;
 
-typedef struct
+static void BgwPoolMainLoop(BgwPool* pool)
 {
-    BgwPoolConstructor constructor;
-    int id;
-} BgwPoolExecutorCtx;
-
-static void BgwPoolMainLoop(Datum arg)
-{
-    BgwPoolExecutorCtx* ctx = (BgwPoolExecutorCtx*)arg;
-    int id = ctx->id;
-    BgwPool* pool = ctx->constructor();
     int size;
     void* work;
 
@@ -58,7 +51,7 @@ static void BgwPoolMainLoop(Datum arg)
 			pool->lastPeakTime = 0;
         }
         SpinLockRelease(&pool->lock);
-        pool->executor(id, work, size);
+        pool->executor(work, size);
         free(work);
         SpinLockAcquire(&pool->lock);
         pool->active -= 1;
@@ -84,6 +77,7 @@ void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname,  c
     pool->pending = 0;
 	pool->nWorkers = nWorkers;
 	pool->lastPeakTime = 0;
+	pool->lastDynamicWorkerStartTime = 0;
 	strncpy(pool->dbname, dbname, MAX_DBNAME_LEN);
 	strncpy(pool->dbuser, dbuser, MAX_DBUSER_LEN);
 }
@@ -93,6 +87,17 @@ timestamp_t BgwGetLastPeekTime(BgwPool* pool)
 	return pool->lastPeakTime;
 }
 
+static void BgwPoolStaticWorkerMainLoop(Datum arg)
+{
+	BgwPoolConstructor constructor = (BgwPoolConstructor)DatumGetPointer(arg);
+    BgwPoolMainLoop(constructor());
+}
+
+static void BgwPoolDynamicWorkerMainLoop(Datum arg)
+{
+    BgwPoolMainLoop((BgwPool*)DatumGetPointer(arg));
+}
+
 void BgwPoolStart(int nWorkers, BgwPoolConstructor constructor)
 {
     int i;
@@ -101,15 +106,12 @@ void BgwPoolStart(int nWorkers, BgwPoolConstructor constructor)
 	MemSet(&worker, 0, sizeof(BackgroundWorker));
     worker.bgw_flags = BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION;
 	worker.bgw_start_time = BgWorkerStart_ConsistentState;
-	worker.bgw_main = BgwPoolMainLoop;
+	worker.bgw_main = BgwPoolStaticWorkerMainLoop;
 	worker.bgw_restart_time = MULTIMASTER_BGW_RESTART_TIMEOUT;
 
     for (i = 0; i < nWorkers; i++) { 
-        BgwPoolExecutorCtx* ctx = (BgwPoolExecutorCtx*)malloc(sizeof(BgwPoolExecutorCtx));
         snprintf(worker.bgw_name, BGW_MAXLEN, "bgw_pool_worker_%d", i+1);
-        ctx->id = i;
-        ctx->constructor = constructor;
-        worker.bgw_main_arg = (Datum)ctx;
+        worker.bgw_main_arg = PointerGetDatum(constructor);
         RegisterBackgroundWorker(&worker);
     }
 }
@@ -124,14 +126,36 @@ size_t BgwPoolGetQueueSize(BgwPool* pool)
 }
 
 
+static void BgwStartExtraWorker(BgwPool* pool)
+{
+	if (pool->nWorkers < MtmMaxWorkers) { 
+		timestamp_t now = MtmGetSystemTime();
+		if (pool->lastDynamicWorkerStartTime + MULTIMASTER_BGW_RESTART_TIMEOUT*USECS_PER_SEC < now) { 
+			BackgroundWorker worker;
+			BackgroundWorkerHandle* handle;
+			MemSet(&worker, 0, sizeof(BackgroundWorker));
+			worker.bgw_flags = BGWORKER_SHMEM_ACCESS |  BGWORKER_BACKEND_DATABASE_CONNECTION;
+			worker.bgw_start_time = BgWorkerStart_ConsistentState;
+			worker.bgw_main = BgwPoolDynamicWorkerMainLoop;
+			worker.bgw_restart_time = MULTIMASTER_BGW_RESTART_TIMEOUT;
+			snprintf(worker.bgw_name, BGW_MAXLEN, "bgw_pool_dynworker_%d", (int)++pool->nWorkers);
+			worker.bgw_main_arg = PointerGetDatum(pool);
+			pool->lastDynamicWorkerStartTime = now;
+			if (!RegisterDynamicBackgroundWorker(&worker, &handle)) { 
+				elog(WARNING, "Failed to start dynamic background worker");
+			}
+		}
+	}
+}
+
 void BgwPoolExecute(BgwPool* pool, void* work, size_t size)
 {
     if (size+4 > pool->size) {
 		/* 
 		 * Size of work is larger than size of shared buffer: 
 		 * run it immediately
 		 */
-		pool->executor(0, work, size);
+		pool->executor(work, size);
 		return;
 	}
 
@@ -149,6 +173,9 @@ void BgwPoolExecute(BgwPool* pool, void* work, size_t size)
             SpinLockAcquire(&pool->lock);
         } else {
             pool->pending += 1;
+			if (pool->active == pool->nWorkers) { 
+				BgwStartExtraWorker(pool);				
+			}
 			if (pool->lastPeakTime == 0 && pool->active == pool->nWorkers && pool->pending != 0) {
 				pool->lastPeakTime = MtmGetSystemTime();
 			}

bgwpool.h

@@ -5,7 +5,7 @@
 #include "storage/spin.h"
 #include "storage/pg_sema.h"
 
-typedef void(*BgwPoolExecutor)(int id, void* work, size_t size);
+typedef void(*BgwPoolExecutor)(void* work, size_t size);
 
 typedef uint64 timestamp_t;
 
@@ -16,7 +16,8 @@ typedef uint64 timestamp_t;
 extern timestamp_t MtmGetSystemTime(void);   /* non-adjusted current system time */
 extern timestamp_t MtmGetCurrentTime(void);  /* adjusted current system time */
 
-extern bool  MtmIsLogicalReceiver;
+extern bool MtmIsLogicalReceiver;
+extern int  MtmMaxWorkers;
 
 typedef struct
 {
@@ -31,6 +32,7 @@ typedef struct
     size_t pending;
 	size_t nWorkers;
 	time_t lastPeakTime;
+	timestamp_t lastDynamicWorkerStartTime;
     bool   producerBlocked;
     char   dbname[MAX_DBNAME_LEN];
 	char   dbuser[MAX_DBUSER_LEN];

multimaster.c

@@ -2370,7 +2370,7 @@ _PG_init(void)
 
 	DefineCustomIntVariable(
 		"multimaster.workers",
-		"Number of multimaster executor workers per node",
+		"Number of multimaster executor workers",
 		NULL,
 		&MtmWorkers,
 		8,
@@ -2383,6 +2383,21 @@ _PG_init(void)
 		NULL
 	);
 
+	DefineCustomIntVariable(
+		"multimaster.max_workers",
+		"Maximal number of multimaster dynamic executor workers",
+		NULL,
+		&MtmMaxWorkers,
+		100,
+		0,
+		INT_MAX,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
 	DefineCustomIntVariable(
 		"multimaster.vacuum_delay",
 		"Minimal age of records which can be vacuumed (seconds)",
@@ -2919,6 +2934,7 @@ mtm_get_snapshot(PG_FUNCTION_ARGS)
 	PG_RETURN_INT64(MtmTx.snapshot);
 }
 
+
 Datum
 mtm_get_last_csn(PG_FUNCTION_ARGS)
 {
@@ -3785,7 +3801,7 @@ void MtmExecute(void* work, int size)
 {
 	if (Mtm->status == MTM_RECOVERY) { 
 		/* During recovery apply changes sequentially to preserve commit order */
-		MtmExecutor(0, work, size);
+		MtmExecutor(work, size);
 	} else { 
 		BgwPoolExecute(&Mtm->pool, work, size);
 	}

multimaster.h

@@ -301,7 +301,7 @@ extern void  MtmJoinTransaction(GlobalTransactionId* gtid, csn_t snapshot);
 extern void  MtmReceiverStarted(int nodeId);
 extern MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shutdown);
 extern void  MtmExecute(void* work, int size);
-extern void  MtmExecutor(int id, void* work, size_t size);
+extern void  MtmExecutor(void* work, size_t size);
 extern void  MtmSend2PCMessage(MtmTransState* ts, MtmMessageCode cmd);
 extern void  MtmSendMessage(MtmArbiterMessage* msg);
 extern void  MtmAdjustSubtransactions(MtmTransState* ts);

pglogical_apply.c

@@ -953,7 +953,7 @@ process_remote_delete(StringInfo s, Relation rel)
 
 static MemoryContext ApplyContext;
 
-void MtmExecutor(int id, void* work, size_t size)
+void MtmExecutor(void* work, size_t size)
 {
     StringInfoData s;
     Relation rel = NULL;

pglogical_receiver.c

@@ -530,7 +530,7 @@ pglogical_receiver_main(Datum main_arg)
 						} else { 
 							if (MtmPreserveCommitOrder && buf.used == rc - hdr_len) {
 								/* Perform commit-prepared and rollback-prepared requested directly in receiver */
-								MtmExecutor(nodeId, buf.data, buf.used);
+								MtmExecutor(buf.data, buf.used);
 							} else {
 								MtmExecute(buf.data, buf.used);
 							}