Fix bug in deadlock detection algorithm

Author: knizhnik

contrib/multimaster/bgwpool.c

@@ -12,20 +12,23 @@
 
 typedef struct
 {
-    BgwPool* pool;
+    BgwPoolConstructor constructor;
     int id;
-} BgwExecutorCtx;
+} BgwPoolExecutorCtx;
 
 static void BgwPoolMainLoop(Datum arg)
 {
-    BgwExecutorCtx* ctx = (BgwExecutorCtx*)arg;
+    BgwPoolExecutorCtx* ctx = (BgwPoolExecutorCtx*)arg;
     int id = ctx->id;
-    BgwPool* pool = ctx->pool;
+    BgwPool* pool = ctx->constructor();
     int size;
     void* work;
 
+    BackgroundWorkerUnblockSignals();
 	BackgroundWorkerInitializeConnection(pool->dbname, NULL);
 
+    elog(WARNING, "Start background worker %d", id);
+
     while(true) { 
         PGSemaphoreLock(&pool->available);
         SpinLockAcquire(&pool->lock);
@@ -52,11 +55,9 @@ static void BgwPoolMainLoop(Datum arg)
     }
 }
 
-BgwPool* BgwPoolCreate(BgwExecutor executor, char const* dbname, size_t queueSize, int nWorkers)
+void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname, size_t queueSize)
 {
-    int i;
-	BackgroundWorker worker;
-    BgwPool* pool = (BgwPool*)ShmemAlloc(queueSize + sizeof(BgwPool));
+    pool->queue = (char*)ShmemAlloc(queueSize);
     pool->executor = executor;
     PGSemaphoreCreate(&pool->available);
     PGSemaphoreCreate(&pool->overflow);
@@ -68,22 +69,27 @@ BgwPool* BgwPoolCreate(BgwExecutor executor, char const* dbname, size_t queueSiz
     pool->tail = 0;
     pool->size = queueSize;
     strcpy(pool->dbname, dbname);
+}
+
+void BgwPoolStart(int nWorkers, BgwPoolConstructor constructor)
+{
+    int i;
+	BackgroundWorker worker;
 
 	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_restart_time = 10; /* Wait 10 seconds for restart before crash */
-
+    
     for (i = 0; i < nWorkers; i++) { 
-        BgwExecutorCtx* ctx = (BgwExecutorCtx*)malloc(sizeof(BgwExecutorCtx));
+        BgwPoolExecutorCtx* ctx = (BgwPoolExecutorCtx*)malloc(sizeof(BgwPoolExecutorCtx));
         snprintf(worker.bgw_name, BGW_MAXLEN, "bgw_pool_worker_%d", i+1);
         ctx->id = i;
-        ctx->pool = pool;
+        ctx->constructor = constructor;
         worker.bgw_main_arg = (Datum)ctx;
         RegisterBackgroundWorker(&worker);
     }
-    return pool;
 }
 
 void BgwPoolExecute(BgwPool* pool, void* work, size_t size)

contrib/multimaster/bgwpool.h

@@ -1,25 +1,33 @@
 #ifndef __BGWPOOL_H__
 #define __BGWPOOL_H__
 
-typedef void(*BgwExecutor)(int id, void* work, size_t size);
+#include "storage/s_lock.h"
+#include "storage/spin.h"
+#include "storage/pg_sema.h"
+
+typedef void(*BgwPoolExecutor)(int id, void* work, size_t size);
 
 #define MAX_DBNAME_LEN 30
 
 typedef struct
 {
-    BgwExecutor executor;
+    BgwPoolExecutor executor;
     volatile slock_t lock;
     PGSemaphoreData available;
     PGSemaphoreData overflow;
     size_t head;
     size_t tail;
     size_t size;
-    bool producerBlocked;
-    char dbname[MAX_DBNAME_LEN];
-    char queue[1];
+    bool   producerBlocked;
+    char   dbname[MAX_DBNAME_LEN];
+    char*  queue;
 } BgwPool;
 
-extern BgwPool* BgwPoolCreate(BgwExecutor executor, char const* dbname, size_t queueSize, int nWorkers);
+typedef BgwPool*(*BgwPoolConstructor)(void);
+
+extern void BgwPoolStart(int nWorkers, BgwPoolConstructor constructor);
+
+extern void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname, size_t queueSize);
 
 extern void BgwPoolExecute(BgwPool* pool, void* work, size_t size);
 

contrib/multimaster/dtmd/src/ddd.c

@@ -116,7 +116,7 @@ void addSubgraph(Graph* graph, nodeid_t node_id, xid_t* xids, int n_xids)
         if (--e->dst->nIncomingEdges == 0 && l2_list_is_empty(&e->dst->outgoingEdges)) {
             freeVertex(graph, e->dst);
         }
-        if (e->src->nIncomingEdges == 0 && l2_list_is_empty(&e->src->outgoingEdges)) {
+        if (e->dst != e->src && e->src->nIncomingEdges == 0 && l2_list_is_empty(&e->src->outgoingEdges)) {
             freeVertex(graph, e->src);
         }
         freeEdge(graph, e);

contrib/multimaster/multimaster.c

@@ -61,7 +61,7 @@ typedef struct
     int    nNodes;
     pg_atomic_uint32 nReceivers;
     bool initialized;
-    BgwPool* pool;
+    BgwPool pool;
 } DtmState;
 
 typedef struct
@@ -104,6 +104,7 @@ static void DtmBackgroundWorker(Datum arg);
 
 static void MMMarkTransAsLocal(TransactionId xid);
 static void MMExecutor(int id, void* work, size_t size);
+static BgwPool* MMPoolConstructor(void);
 
 static shmem_startup_hook_type prev_shmem_startup_hook;
 
@@ -716,7 +717,7 @@ static void DtmInitialize()
         dtm->nNodes = MMNodes;
         pg_atomic_write_u32(&dtm->nReceivers, 0);
         dtm->initialized = false;
-        dtm->pool = BgwPoolCreate(MMExecutor, MMDatabaseName, MMQueueSize, MMWorkers);
+        BgwPoolInit(&dtm->pool, MMExecutor, MMDatabaseName, MMQueueSize);
 		RegisterXactCallback(DtmXactCallback, NULL);
 	}
 	LWLockRelease(AddinShmemInitLock);
@@ -953,6 +954,7 @@ _PG_init(void)
     if (MMNodes < 2) { 
         elog(ERROR, "Multimaster should have at least two nodes");
     }
+    BgwPoolStart(MMWorkers, MMPoolConstructor);
 
 	if (DtmBufferSize != 0)
 	{
@@ -1203,7 +1205,6 @@ static void MMExecutor(int id, void* work, size_t size)
         SPI_finish();
         PopActiveSnapshot();
         if (rc != SPI_OK_INSERT && rc != SPI_OK_UPDATE && rc != SPI_OK_DELETE) {
-            FlushErrorState();
             ereport(LOG, (errmsg("Executor %d: failed to apply transaction %u",
                                  id, xid)));
             AbortCurrentTransaction();
@@ -1213,6 +1214,7 @@ static void MMExecutor(int id, void* work, size_t size)
     }
     PG_CATCH();
     {
+        FlushErrorState();
         if (rc == SPI_ERROR_TRANSACTION) {
             SPI_finish();
             PopActiveSnapshot();
@@ -1224,6 +1226,10 @@ static void MMExecutor(int id, void* work, size_t size)
 
 extern void MMExecute(void* work, int size)
 {
-    BgwPoolExecute(dtm->pool, work, size);
+    BgwPoolExecute(&dtm->pool, work, size);
 }
 
+static BgwPool* MMPoolConstructor(void)
+{
+    return &dtm->pool;
+}

contrib/pg_dtm/dtmd/src/ddd.c

@@ -116,7 +116,7 @@ void addSubgraph(Graph* graph, nodeid_t node_id, xid_t* xids, int n_xids)
         if (--e->dst->nIncomingEdges == 0 && l2_list_is_empty(&e->dst->outgoingEdges)) {
             freeVertex(graph, e->dst);
         }
-        if (e->src->nIncomingEdges == 0 && l2_list_is_empty(&e->src->outgoingEdges)) {
+        if (e->dst != e->src && e->src->nIncomingEdges == 0 && l2_list_is_empty(&e->src->outgoingEdges)) {
             freeVertex(graph, e->src);
         }
         freeEdge(graph, e);