new semaphore api

Author: kelvich

arbiter.c

@@ -340,7 +340,7 @@ static void MtmScheduleHeartbeat()
 		enable_timeout_after(heartbeat_timer, MtmHeartbeatSendTimeout);
 		send_heartbeat = true;
 	}
-	PGSemaphoreUnlock(&Mtm->sendSemaphore);
+	PGSemaphoreUnlock(Mtm->sendSemaphore);
 }
 
 static void MtmSendHeartbeat()
@@ -741,7 +741,7 @@ static void MtmSender(Datum arg)
 
 	while (!stop) {
 		MtmMessageQueue *curr, *next;		
-		PGSemaphoreLock(&Mtm->sendSemaphore);
+		PGSemaphoreLock(Mtm->sendSemaphore);
 		CHECK_FOR_INTERRUPTS();
 
 		if (ConfigReloadPending)

bgwpool.c

@@ -59,10 +59,10 @@ static void BgwPoolMainLoop(BgwPool* pool)
 			ProcessConfigFile(PGC_SIGHUP);
 		}
 
-        PGSemaphoreLock(&pool->available);
+		PGSemaphoreLock(pool->available);
         SpinLockAcquire(&pool->lock);
 		if (pool->shutdown) { 	
-			PGSemaphoreUnlock(&pool->available);
+			PGSemaphoreUnlock(pool->available);
 			break;
 		}
         size = *(int*)&pool->queue[pool->head];
@@ -85,7 +85,7 @@ static void BgwPoolMainLoop(BgwPool* pool)
         }
         if (pool->producerBlocked) {
             pool->producerBlocked = false;
-            PGSemaphoreUnlock(&pool->overflow);
+			PGSemaphoreUnlock(pool->overflow);
 			pool->lastPeakTime = 0;
         }
         SpinLockRelease(&pool->lock);
@@ -108,10 +108,10 @@ void BgwPoolInit(BgwPool* pool, BgwPoolExecutor executor, char const* dbname,  c
 		elog(PANIC, "Failed to allocate memory for background workers pool: %lld bytes requested", (long64)queueSize);
 	}
     pool->executor = executor;
-    PGSemaphoreCreate(&pool->available);
-    PGSemaphoreCreate(&pool->overflow);
-    PGSemaphoreReset(&pool->available);
-    PGSemaphoreReset(&pool->overflow);
+	pool->available = PGSemaphoreCreate();
+	pool->overflow = PGSemaphoreCreate();
+	PGSemaphoreReset(pool->available);
+	PGSemaphoreReset(pool->overflow);
     SpinLockInit(&pool->lock);
 	pool->shutdown = false;
     pool->producerBlocked = false;
@@ -215,7 +215,7 @@ void BgwPoolExecute(BgwPool* pool, void* work, size_t size)
 			}
 			pool->producerBlocked = true;
             SpinLockRelease(&pool->lock);
-            PGSemaphoreLock(&pool->overflow);
+			PGSemaphoreLock(pool->overflow);
             SpinLockAcquire(&pool->lock);
         } else {
             pool->pending += 1;
@@ -236,7 +236,7 @@ void BgwPoolExecute(BgwPool* pool, void* work, size_t size)
             if (pool->tail == pool->size) {
                 pool->tail = 0;
             }
-            PGSemaphoreUnlock(&pool->available);
+			PGSemaphoreUnlock(pool->available);
             break;
         }
     }
@@ -248,6 +248,6 @@ void BgwPoolStop(BgwPool* pool)
     SpinLockAcquire(&pool->lock);
 	pool->shutdown = true;
     SpinLockRelease(&pool->lock);            
-	PGSemaphoreUnlock(&pool->available);
-	PGSemaphoreUnlock(&pool->overflow);
+	PGSemaphoreUnlock(pool->available);
+	PGSemaphoreUnlock(pool->overflow);
 }

bgwpool.h

@@ -25,8 +25,8 @@ typedef struct
 {
     BgwPoolExecutor executor;
     volatile slock_t lock;
-    PGSemaphoreData available;
-    PGSemaphoreData overflow;
+	PGSemaphore available;
+	PGSemaphore overflow;
     size_t head;
     size_t tail;
     size_t size;

multimaster.c

@@ -1694,7 +1694,7 @@ void MtmSendMessage(MtmArbiterMessage* msg)
 		Mtm->sendQueue = mq;
 		if (sendQueue == NULL) {
 			/* signal semaphore only once for the whole list */
-			PGSemaphoreUnlock(&Mtm->sendSemaphore);
+			PGSemaphoreUnlock(Mtm->sendSemaphore);
 		}
 	}
 	SpinLockRelease(&Mtm->queueSpinlock);
@@ -2547,8 +2547,8 @@ static void MtmInitialize()
 		Mtm->nodes[MtmNodeId-1].originId = DoNotReplicateId;
 		/* All transaction originated from the current node should be ignored during recovery */
 		Mtm->nodes[MtmNodeId-1].restartLSN = (lsn_t)PG_UINT64_MAX;
-		PGSemaphoreCreate(&Mtm->sendSemaphore);
-		PGSemaphoreReset(&Mtm->sendSemaphore);
+		Mtm->sendSemaphore = PGSemaphoreCreate();
+		PGSemaphoreReset(Mtm->sendSemaphore);
 		SpinLockInit(&Mtm->queueSpinlock);
 		BgwPoolInit(&Mtm->pool, MtmExecutor, MtmDatabaseName, MtmDatabaseUser, MtmQueueSize, MtmWorkers);
 		RegisterXactCallback(MtmXactCallback, NULL);

multimaster.h

@@ -292,7 +292,7 @@ typedef struct
 	MtmNodeStatus status;              /* Status of this node */
 	int recoverySlot;                  /* NodeId of recovery slot or 0 if none */
 	volatile slock_t queueSpinlock;    /* spinlock used to protect sender queue */
-	PGSemaphoreData sendSemaphore;     /* semaphore used to notify mtm-sender about new responses to coordinator */
+	PGSemaphore sendSemaphore;     /* semaphore used to notify mtm-sender about new responses to coordinator */
 	LWLockPadded *locks;               /* multimaster lock tranche */
 	TransactionId oldestXid;           /* XID of oldest transaction visible by any active transaction (local or global) */
 	nodemask_t disabledNodeMask;       /* Bitmask of disabled nodes */